
构建LLM智能体
LLM智能体通过工具、记忆和规划扩展语言模型,以处理复杂的多步任务。
使用LangGraph的ReAct智能体
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated
import operator
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
model = ChatOpenAI(model="gpt-4o", temperature=0)
def should_continue(state):
return "tools" if state["messages"][-1].tool_calls else END
def call_model(state):
return {"messages": [model_with_tools.invoke(state["messages"])]}
def build_agent(tools):
global model_with_tools
model_with_tools = model.bind_tools(tools)
tool_node = ToolNode(tools)
wf = StateGraph(AgentState)
wf.add_node("agent", call_model)
wf.add_node("tools", tool_node)
wf.set_entry_point("agent")
wf.add_conditional_edges("agent", should_continue)
wf.add_edge("tools", "agent")
return wf.compile()
自定义工具
from langchain_core.tools import tool
@tool
def search_web(query: str) -> str:
"""Search the web for current information."""
from langchain_community.tools import DuckDuckGoSearchRun
return DuckDuckGoSearchRun().run(query)
@tool
def execute_python(code: str) -> str:
"""Execute Python code and return result."""
import subprocess
r = subprocess.run(["python3", "-c", code], capture_output=True, text=True, timeout=10)
return r.stdout if r.returncode == 0 else f"Error: {r.stderr}"
智能体记忆
class AgentMemory:
def __init__(self, session_id, vectorstore):
self.session_id = session_id
self.vectorstore = vectorstore
self.short_term = []
def add(self, role, content):
self.short_term.append({"role": role, "content": content})
if len(self.short_term) > 20:
self._archive(self.short_term[:-20])
self.short_term = self.short_term[-20:]
def recall(self, query, k=3):
docs = self.vectorstore.similarity_search(
query, k=k, filter={"session_id": self.session_id}
)
return [d.page_content for d in docs]
生产注意事项
| 关注点 | 解决方案 |
|---|---|
| 无限循环 | 最大迭代次数限制 |
| Token成本 | Token计数 + 预算 |
| 工具失败 | 带退避的重试 |
| 可观测性 | LangSmith追踪 |