AI Agent 架构：自治智能体与 LangChain 框架
AI Agents 是能够自主做出决策、使用工具并逐步解决复杂任务的人工智能系统。本指南将探讨智能体架构及其实现方式。
什么是 Agent？
聊天机器人和智能体的简单对比：
1Chatbot: 2User: "How is the weather today?" 3Bot: "I don't know the weather." (static response) 4 5Agent: 6User: "How is the weather today?" 7Agent: 8 1. "I need to find out the weather." 9 2. Call weather_api.get_weather() 10 3. Interpret the result 11 4. "It's 15°C in Istanbul, partly cloudy."
Agent 组件
1. LLM（大脑）
负责决策和推理。
2. Tools（工具）
与外部世界交互：
- Web 搜索
- API 调用
- 数据库查询
- 代码执行
3. Memory（记忆）
存储对话历史和信息。
4. Planning（规划）
任务分解与策略制定。
ReAct 模式
推理 + 行动循环：
1Thought: What do I need to do? 2Action: Which tool should I use? 3Action Input: Tool parameters 4Observation: Tool result 5... (repeat) 6Final Answer: Final response
ReAct 示例
1Question: "What is Tesla's current stock price and how much has it changed in the last month?" 2 3Thought: I need to find Tesla's stock price. 4Action: stock_price 5Action Input: {"symbol": "TSLA"} 6Observation: Current price: $248.50 7 8Thought: Now I need to calculate the 1-month change. 9Action: stock_history 10Action Input: {"symbol": "TSLA", "period": "1mo"} 11Observation: 1 month ago: $235.20, Change: +5.65% 12 13Thought: I can now answer the question. 14Final Answer: Tesla's current stock price is $248.50. 15It has shown a 5.65% increase in the last month.
使用 LangChain 进行 Agent 开发
基础 Agent
1from langchain.agents import create_react_agent, AgentExecutor 2from langchain.tools import Tool 3from langchain_openai import ChatOpenAI 4from langchain import hub 5 6# LLM 7llm = ChatOpenAI(model="gpt-4-turbo", temperature=0) 8 9# Tools 10def search_web(query: str) -> str: 11 """Searches the web""" 12 # Web search implementation 13 return f"Search results for: {query}" 14 15def calculate(expression: str) -> str: 16 """Performs mathematical calculation""" 17 return str(eval(expression)) 18 19tools = [ 20 Tool(name="search", func=search_web, description="Searches the web"), 21 Tool(name="calculator", func=calculate, description="Performs math calculations"), 22] 23 24# Prompt 25prompt = hub.pull("hwchase17/react") 26 27# Agent 28agent = create_react_agent(llm, tools, prompt) 29agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) 30 31# Run 32result = agent_executor.invoke({"input": "What is the world population in 2024?"})
自定义工具
1from langchain.tools import BaseTool 2from pydantic import BaseModel, Field 3 4class WeatherInput(BaseModel): 5 city: str = Field(description="City name") 6 unit: str = Field(default="celsius", description="Temperature unit") 7 8class WeatherTool(BaseTool): 9 name = "weather" 10 description = "Fetches weather information" 11 args_schema = WeatherInput 12 13 def _run(self, city: str, unit: str = "celsius") -> str: 14 # API call 15 weather_data = fetch_weather(city) 16 return f"{city}: {weather_data['temp']}°{'C' if unit == 'celsius' else 'F'}" 17 18 async def _arun(self, city: str, unit: str = "celsius") -> str: 19 return await async_fetch_weather(city) 20## 内存系统 21 22### Conversation Buffer Memory（对话缓冲记忆） 23 24```python 25from langchain.memory import ConversationBufferMemory 26 27memory = ConversationBufferMemory( 28 memory_key="chat_history", 29 return_messages=True 30) 31 32# Add to Agent 33agent_executor = AgentExecutor( 34 agent=agent, 35 tools=tools, 36 memory=memory, 37 verbose=True 38)
Summary Memory（摘要记忆）
用于长对话的摘要：
1from langchain.memory import ConversationSummaryMemory 2 3memory = ConversationSummaryMemory( 4 llm=llm, 5 memory_key="chat_history" 6)
Vector Store Memory（向量存储记忆）
带语义搜索的长期记忆：
1from langchain.memory import VectorStoreRetrieverMemory 2from langchain.vectorstores import Chroma 3 4vectorstore = Chroma(embedding_function=embeddings) 5retriever = vectorstore.as_retriever(search_kwargs={"k": 5}) 6 7memory = VectorStoreRetrieverMemory(retriever=retriever)
多智能体系统
Supervisor Pattern（监督者模式）
1from langgraph.graph import StateGraph, END 2 3def supervisor_agent(state): 4 """Supervisor distributing tasks""" 5 task = state["task"] 6 7 if "research" in task: 8 return "researcher" 9 elif "write" in task: 10 return "writer" 11 else: 12 return "general" 13 14def researcher_agent(state): 15 """Agent conducting research""" 16 # Research logic 17 return {"research_results": results} 18 19def writer_agent(state): 20 """Agent writing content""" 21 # Writing logic 22 return {"content": content} 23 24# Create Graph 25workflow = StateGraph(AgentState) 26workflow.add_node("supervisor", supervisor_agent) 27workflow.add_node("researcher", researcher_agent) 28workflow.add_node("writer", writer_agent) 29 30workflow.add_conditional_edges("supervisor", route_to_agent) 31workflow.add_edge("researcher", "supervisor") 32workflow.add_edge("writer", END) 33 34app = workflow.compile()
协作式 Agents
1┌─────────────────────────────────────────────────────┐ 2│ Orchestrator │ 3└───────────────────────┬─────────────────────────────┘ 4 │ 5 ┌───────────────┼───────────────┐ 6 │ │ │ 7┌───────▼───────┐ ┌─────▼─────┐ ┌──────▼──────┐ 8│ Researcher │ │ Analyst │ │ Writer │ 9│ Agent │ │ Agent │ │ Agent │ 10└───────────────┘ └───────────┘ └─────────────┘
规划策略
Plan-and-Execute（规划并执行）
1from langchain_experimental.plan_and_execute import ( 2 PlanAndExecute, 3 load_agent_executor, 4 load_chat_planner 5) 6 7planner = load_chat_planner(llm) 8executor = load_agent_executor(llm, tools, verbose=True) 9 10agent = PlanAndExecute(planner=planner, executor=executor) 11 12result = agent.run("Research and create a draft for a blog post")
Task Decomposition（任务分解）
1Main Task: "Analyze products for an e-commerce site" 2 3Sub-Tasks: 41. Research competitor products 52. Perform price comparison 63. Analyze customer reviews 74. Generate report 8 9Each sub-task is handled by an agent. 10## 错误处理与重试 11 12```python 13from langchain.callbacks import get_openai_callback 14 15class RobustAgentExecutor: 16 def __init__(self, agent_executor, max_retries=3): 17 self.executor = agent_executor 18 self.max_retries = max_retries 19 20 def invoke(self, input_dict): 21 for attempt in range(self.max_retries): 22 try: 23 with get_openai_callback() as cb: 24 result = self.executor.invoke(input_dict) 25 26 # Token limit check 27 if cb.total_tokens > 10000: 28 self.summarize_memory() 29 30 return result 31 32 except Exception as e: 33 if attempt == self.max_retries - 1: 34 raise 35 36 # Strategy based on Error 37 if "rate_limit" in str(e): 38 time.sleep(60) 39 elif "context_length" in str(e): 40 self.truncate_context() 41 else: 42 time.sleep(2 ** attempt)
流式与实时处理
1from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler 2 3llm = ChatOpenAI( 4 streaming=True, 5 callbacks=[StreamingStdOutCallbackHandler()] 6) 7 8async def stream_agent_response(query): 9 async for event in agent_executor.astream_events( 10 {"input": query}, 11 version="v1" 12 ): 13 kind = event["event"] 14 if kind == "on_chat_model_stream": 15 print(event["data"]["chunk"].content, end="") 16 elif kind == "on_tool_start": 17 print(f"\n🔧 Using tool: {event['name']}")
生产环境注意事项
日志与监控
1import logging 2 3class AgentLogger: 4 def __init__(self): 5 self.logger = logging.getLogger("agent") 6 7 def log_action(self, action, input_data, output): 8 self.logger.info({ 9 "action": action, 10 "input": input_data, 11 "output_length": len(str(output)), 12 "timestamp": datetime.now().isoformat() 13 }) 14 15 def log_error(self, error, context): 16 self.logger.error({ 17 "error": str(error), 18 "context": context 19 })
成本管理
1def estimate_cost(agent_run): 2 # Cost per token (GPT) 3 input_cost = agent_run.input_tokens * 0.00003 4 output_cost = agent_run.output_tokens * 0.00006 5 6 return input_cost + output_cost 7 8# Budget limit 9MAX_BUDGET_PER_REQUEST = 0.50 # $0.50 10 11if estimate_cost(run) > MAX_BUDGET_PER_REQUEST: 12 raise BudgetExceededError()
总结
AI Agents 使人工智能能够以自主、主动的方式运行。像 LangChain 和 LangGraph 这样的框架让开发可用于生产环境的智能体系统变得更加容易。
在 Veni AI，我们构建企业级智能体解决方案。