正在加载,请稍候…

Building LLM Agents: Tool Use, Memory, and Multi-Step Reasoning

Learn to build production LLM agents with tool calling, persistent memory, planning capabilities, and multi-agent orchestration using LangGraph.

Building LLM Agents

LLM agents extend language models with tools, memory, and planning for complex multi-step tasks.

ReAct Agent with LangGraph

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

Custom Tools

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

Agent Memory

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]

Production Notes

Concern Solution
Infinite loops Max iteration limits
Token costs Token counting + budgets
Tool failures Retry with backoff
Observability LangSmith tracing