ai

2025年构建AI代理：从ChatGPT到Multi-Agent系统

发布: (2026年1月12日 GMT+8 16:56)
6 min read
原文: Dev.to

Source: Dev.to

请提供您希望翻译的具体文本内容（文章正文），我将为您翻译成简体中文，并保留原始的格式、Markdown 语法以及技术术语。谢谢！

什么是 AI 代理？

AI 代理是一种能够感知其环境、做出决策并采取行动以实现特定目标的自主系统。与传统软件不同，AI 代理可以从交互中适应和学习。

AI 代理类型

  • 简单反射代理 – 对当前状态作出反应。
  • 基于模型的代理 – 维护内部状态。
  • 基于目标的代理 – 朝特定目标工作。
  • 基于效用的代理 – 优化最佳结果。
  • 学习型代理 – 随时间提升性能。

构建您的第一个 AI 代理

使用 LangChain

from langchain.agents import create_openai_functions_agent
from langchain.tools import Tool
from langchain_openai import ChatOpenAI

# Define tools
def search_web(query: str) -> str:
    """Search the web for information"""
    # Implementation here
    return f"Results for: {query}"

def calculate(expression: str) -> str:
    """Calculate mathematical expressions"""
    return str(eval(expression))

tools = [
    Tool(name="Search", func=search_web, description="Search the web"),
    Tool(name="Calculator", func=calculate, description="Calculate math")
]

# Create agent
llm = ChatOpenAI(model="gpt-4", temperature=0)
agent = create_openai_functions_agent(llm, tools)

# Run agent
result = agent.run("What's 15% of 1200?")
print(result)

Source: …

高级：多代理系统

多代理系统涉及多个 AI 代理协同工作，每个代理都有其专门的角色。

AutoGen 框架

import autogen

config_list = [{
    "model": "gpt-4",
    "api_key": "your-key"
}]

# Define agents
user_proxy = autogen.UserProxyAgent(
    name="user",
    human_input_mode="TERMINATE",
    max_consecutive_auto_reply=10
)

coder = autogen.AssistantAgent(
    name="coder",
    llm_config={"config_list": config_list},
    system_message="You write Python code to solve tasks."
)

reviewer = autogen.AssistantAgent(
    name="reviewer",
    llm_config={"config_list": config_list},
    system_message="You review code for bugs and improvements."
)

# Group chat
groupchat = autogen.GroupChat(
    agents=[user_proxy, coder, reviewer],
    messages=[],
    max_round=12
)

manager = autogen.GroupChatManager(
    groupchat=groupchat,
    llm_config={"config_list": config_list}
)

# Start conversation
user_proxy.initiate_chat(
    manager,
    message="Build a web scraper for product prices"
)

代理记忆系统

记忆对于上下文感知的代理至关重要：

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings

# Short‑term memory
memory = ConversationBufferMemory()

# Long‑term memory with vector store
vectorstore = FAISS.from_texts(
    ["User likes React", "User prefers TypeScript"],
    OpenAIEmbeddings()
)

# Retrieve relevant memories
relevant = vectorstore.similarity_search("What framework?")
print(relevant)

工具创建用于代理

from langchain.tools import BaseTool
from pydantic import BaseModel, Field

class DatabaseQueryInput(BaseModel):
    query: str = Field(description="SQL query to execute")

class DatabaseTool(BaseTool):
    name = "database"
    description = "Query the database"
    args_schema = DatabaseQueryInput

    def _run(self, query: str) -> str:
        # Execute query safely
        # Return results
        return f"Query results: {query}"

    async def _arun(self, query: str) -> str:
        # Async implementation
        return self._run(query)

ReACT 模式（推理 + 行动）

from langchain.agents import load_tools, initialize_agent, AgentType

tools = load_tools(["serpapi", "llm-math"], llm=llm)

agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

result = agent.run(
    "What's the population of Tokyo and what's 20% of it?"
)

生产注意事项

限流

from functools import wraps
import time

def rate_limit(calls_per_minute=10):
    min_interval = 60.0 / calls_per_minute
    last_called = [0.0]

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            if elapsed < min_interval:
                time.sleep(min_interval - elapsed)
            last_called[0] = time.time()
            return func(*args, **kwargs)
        return wrapper
    return decorator

@rate_limit(calls_per_minute=5)
def call_llm(prompt):
    return llm.invoke(prompt)

错误处理

from tenacity import retry, stop_after_attempt, wait_exponential
import logging

logger = logging.getLogger(__name__)

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10)
)
def resilient_agent_call(agent, input_text):
    try:
        return agent.run(input_text)
    except Exception as e:
        logger.error(f"Agent error: {e}")
        raise

成本监控

from langchain.callbacks import get_openai_callback

with get_openai_callback() as cb:
    result = agent.run("Complex query")
    print(f"Total Tokens: {cb.total_tokens}")
    print(f"Total Cost: ${cb.total_cost}")

实际案例

客户支持代理

tools = [
    search_knowledge_base,
    create_ticket,
    escalate_to_human,
    check_order_status,
]

support_agent = create_agent(
    tools=tools,
    system_message="You're a helpful customer support agent",
)

代码审查代理

code_reviewer = create_agent(
    tools=[analyze_code, suggest_improvements, check_security],
    system_message="Review code for bugs, performance, and security",
)

研究助理

research_agent = create_agent(
    tools=[search_papers, summarize_content, cite_sources],
    system_message="Help with academic research",
)

未来趋势

  • Agentic Workflows – 代理链协同工作
  • Self‑Healing Systems – 能自我修复的系统
  • Hybrid Intelligence – 人类与代理的协同智能
  • Specialized Agent Marketplaces – 针对特定任务的专用代理市场
  • Edge AI Agents – 本地运行的 AI 代理，保障隐私

最佳实践

  • 从简单开始 – 采用单一功能的代理
  • 明确边界 – 定义代理可以和不可以做的事情
  • 人工监督 – 对关键决策保持人工参与
  • 监控性能 – 跟踪成功率和成本
  • 版本控制 – 跟踪代理提示和配置
  • 广泛测试 – 边缘情况可能导致意外行为

结论

AI 代理正在改变我们构建软件的方式。无论您是在创建一个简单的聊天机器人还是一个复杂的多代理系统，关键是从小做起，并根据真实使用情况进行迭代。

软件开发的未来将越来越多地涉及编排 AI 代理，以自主处理复杂任务。

您正在构建哪些 AI 代理？分享您的经验！

探索在生产系统中集成 AI。关注获取更多前沿技术洞察！

Back to Blog

相关文章

阅读更多 »