正在加载,请稍候…

构建LLM智能体:工具使用、记忆与多步推理

学习使用LangGraph构建生产级LLM智能体,涵盖工具调用、持久化记忆、规划能力及多智能体编排。

构建LLM智能体:工具使用、记忆与多步推理

构建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()

构建LLM智能体:工具使用、记忆与多步推理示意图

自定义工具

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}"

构建LLM智能体:工具使用、记忆与多步推理示意图

智能体记忆

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]

构建LLM智能体:工具使用、记忆与多步推理示意图

生产注意事项

关注点 解决方案
无限循环 最大迭代次数限制
Token成本 Token计数 + 预算
工具失败 带退避的重试
可观测性 LangSmith追踪