第 1 章 从 LLM 到 Agent：基础蓝图

代码示例

关于本 notebook

本 notebook 是一份紧凑的端到端入门指南：把一次普通的 LLM 调用，变成一个小而真实的 Agent——能够使用工具、跨轮次保持状态，并追踪自己的推理流程。

它展示了什么

• 无状态 LLM 调用：用 langchain-openai 热身。

• 循环中的工具使用：两个简单的工具

  1. internet_search 通过 SerpAPI 获取最新信息
  2. calculator 由 numexpr 驱动，做快速数学计算。

• 一个最小的 LangGraph Agent：

  • 用 add_messages 累积 messages 的类型化 state
  • 绑定工具、由 ToolNode 执行调用的 llm 节点
  • 决定何时调用工具、何时停止的路由器
  • 通过 MemorySaver 的内存 checkpoint
  • 用独立的 thread 分支对话并对比记忆。

为什么用这些示例

1. 你会看到处理工具调用的最简循环——不依赖任何框架状态。
2. 然后你会看到用 LangGraph 把同样的想法做对：干净的路由、checkpoint 和 thread 级记忆。
3. 你会得到追踪工具，用来打印节点更新和简短的状态快照，这让调试和教学都容易得多。

如何运行

• 从 .env 加载密钥，或用 %env 设置。如果缺失，notebook 会交互式询问。
• 调用 gpt-5-mini，或任何你喜欢的兼容 OAI API 的 LLM，先快速检查一次，再切换到绑定工具的 gpt-4o。
• 运行一个两步任务：
  • 用 internet_search 获取纽约市当前气温
  • 用 calculator 计算该温度的平方。
• 在 LangGraph 应用里重复这个任务，然后分支一条新的 thread，把温度换算成华氏度并同时报告两者。 你会看到每个节点的更新，以及每条 thread 最终的记忆快照。

值得注意的关键点

• 把工具绑定到模型，让模型自己决定何时调用。
• 一个最小路由器，检查 tool_calls，要么转到 ToolNode，要么结束。
• Thread id：用于独立的记忆和可复现的调试。
• 尽量确定性的设置 temperature=0、显式的 recursion_limit，以及简短格式的 prompt，以产生一致的输出。

替换与扩展

• 你可以把 SerpAPIWrapper 换成任何返回文本的搜索工具。
• 添加你自己的工具，放进 tools 和 tool_map。
• 如果你想要长期会话，把 MemorySaver 换成持久化的 checkpointer。
• 如果你想要更大的图，添加规划、校验或护栏（guardrail）节点。

要求与注意事项

• 你需要一个可用的 OPENAI_API_KEY 和一个 SerpAPIWrapper。
• 联网搜索结果会变化。报告的温度和摘要会随时间和来源而不同。
• 工具调用会计入 token 和外部 API 用量。留意成本。

依赖

!pip install -q langgraph==0.6.7 langchain-openai==0.3.33 python-dotenv==1.1.1 langchain_community google-search-results

API 导入

from dotenv import load_dotenv
import os

导入

# Standard library
import os
import math
import json
import numexpr
from typing import List, Dict, Any, TypedDict, Annotated

# LangChain core
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_community.utilities import SerpAPIWrapper

# LangGraph
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.memory import MemorySaver

# --- API Key Setup ---
# Option 1 (preferred): create a `.env` file in your project folder with:
# OPENAI_API_KEY=your_openai_key_here
# SERPAPI_API_KEY=your_serpapi_key_here
#
# Option 2: set it directly in the notebook with magic:
# %env OPENAI_API_KEY=your_openai_key_here
# %env SERPAPI_API_KEY=your_serpapi_key_here

from dotenv import load_dotenv
import os

# Load from .env if available
load_dotenv()

OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
serp_api_key = os.getenv("SERPAPI_API_KEY")

# Fallback: ask if still missing
if not OPENAI_API_KEY:
    print("⚠️ OPENAI_API_KEY not found. You can set it with `%env` in the notebook or enter it below.")
    OPENAI_API_KEY = input("Enter your OPENAI_API_KEY: ").strip()

if not serp_api_key:
    print("⚠️ SERPAPI_API_KEY not found. You can set it with `%env` in the notebook or enter it below.")
    serp_api_key = input("Enter your SERPAPI_API_KEY: ").strip()

print("✅ API keys loaded successfully!")

第一个示例：用 LangChain 做无状态 LLM 调用

---

设置 LLM

llm = ChatOpenAI(model="gpt-5-mini")
response = llm.invoke("What are AI agents?")
print(response.content)

Short answer
An AI agent is a software (or embodied) system that perceives its environment, makes decisions, and takes actions to achieve goals — usually with some degree of autonomy and adaptability.

Key characteristics
- Perception: senses inputs (camera, mic, sensors, API data, user text).
- Decision-making: chooses actions based on goals, models, or learned policies.
- Action: affects the environment (move a robot, send a message, place a trade).
- Autonomy: operates without requiring step-by-step human control.
- Goal-directedness: usually tries to maximize a reward or satisfy objectives.

Common types and examples
- Reactive agents: map inputs directly to actions (simple controllers, thermostats).
- Deliberative/planning agents: build internal models and plan ahead (robot path planners).
- Learning agents: improve with experience (reinforcement learning agents, recommendation systems).
- Hybrid agents: combine rules, planning, and learning.
- Embodied agents: robots, autonomous vehicles.
- Software agents: chatbots, virtual assistants, web crawlers, trading bots.
- Multi-agent systems: multiple agents interacting or negotiating (simulation, swarm robotics).

Typical components
- Sensors/input layer (perception)
- World model or memory (state, beliefs)
- Decision module (rules, planner, policy, or LLM)
- Action/output layer (actuators, APIs)
- Learning/updating mechanism (optional)

How they’re built
- Rule-based systems: explicit rules and logic.
- Model-based planners: search and optimization over possible actions.
- Machine learning: supervised or reinforcement learning to derive policies.
- LLM-based agents: language models orchestrate tools, planning, and memory for complex tasks.

Applications
- Personal assistants (scheduling, email triage)
- Autonomous vehicles and drones
- Industrial automation and robotics
- Customer support chatbots
- Finance: algorithmic trading and portfolio management
- Healthcare diagnostics assistance
- Simulation and gaming (NPCs, training sims)

Limitations and risks
- Brittleness outside training or design conditions
- Erroneous or biased outputs from bad data or objectives
- Safety hazards for physical agents (collisions, damage)
- Security and adversarial manipulation
- Alignment and ethical concerns for goal-setting and autonomy

Good practices
- Define clear objectives and constraints
- Test extensively in realistic scenarios and edge cases
- Provide human oversight and fail-safe controls
- Monitor, log, and update models based on performance
- Consider transparency, privacy, and ethical impacts

If you want, I can:
- Give specific examples (e.g., how a Roomba vs. ChatGPT-like agent is built).
- Explain architectures (reactive vs. deliberative vs. memory-based).
- Suggest tools and libraries for building agents (RL frameworks, agent platforms).

为无状态运行定义工具

工具

@tool("internet_search")
def internet_search(query: str) -> str:
    """Search Google via SerpAPI for up to date information."""
    serp_api_key = os.environ["SERPAPI_API_KEY"]
    params = {"engine": "google", "gl": "us", "hl": "en"}
    search = SerpAPIWrapper(params=params, serpapi_api_key=serp_api_key)
    return search.run(query)

@tool("calculator")
def calculator(expression: str) -> str:
    """Evaluate a single line mathematical expression with numexpr."""
    local_dict = {"pi": math.pi, "e": math.e}
    out = numexpr.evaluate(
        expression.strip(),
        global_dict={},
        local_dict=local_dict,
    )
    return str(out)

tools = [internet_search, calculator]

tool_map: Dict[str, Any] = {t.name: t for t in tools}

模型绑定工具

llm = ChatOpenAI(model="gpt-4o").bind_tools(tools, tool_choice="any")

无状态单次运行的工具循环

def run_once(prompt: str, max_steps: int = 4) -> str:
    messages = [HumanMessage(content=prompt)]
    for _ in range(max_steps):
        ai: AIMessage = llm.invoke(messages)
        messages.append(ai)

        calls = getattr(ai, "tool_calls", None) or []
        if not calls:
            break

        for call in calls:
            name = call["name"]
            args = call.get("args", {})
            result = tool_map[name].invoke(args)
            messages.append(ToolMessage(
                content=str(result),
                name=name,
                tool_call_id=call["id"]
            ))
    return messages[-1].content

print(run_once("""Two step task.

Step 1: Use internet_search to get the current air temperature in New York City today. Show the exact query you used, the top source title and snippet, and extract a numeric temperature in Celsius. Return this temperature as feedback for Step 2.

Step 2: Using the Celsius value from Step 1, compute its square with calculator. Show the exact expression you used and the numeric result.

Important: Give a short final answer in this format:
Current temperature:
Square of current temperature:"""))

工具

@tool("internet_search")
def internet_search(query: str) -> str:
    """Search Google via SerpAPI for up to date information."""
    serp_api_key = os.environ["SERPAPI_API_KEY"]
    params = {"engine": "google", "gl": "us", "hl": "en"}
    search = SerpAPIWrapper(params=params, serpapi_api_key=serp_api_key)
    return search.run(query)

@tool("calculator")
def calculator(expression: str) -> str:
    """Evaluate a single line mathematical expression with numexpr."""
    local_dict = {"pi": math.pi, "e": math.e}
    out = numexpr.evaluate(
        expression.strip(),
        global_dict={},
        local_dict=local_dict,
    )
    return str(out)

tools = [internet_search, calculator]
tool_map: Dict[str, Any] = {t.name: t for t in tools}

模型绑定工具

llm = ChatOpenAI(model="gpt-4o", temperature=0, max_tokens=800).bind_tools(tools, tool_choice="auto")

最小工具循环（无状态）

def run_once(prompt: str, max_steps: int = 8) -> str:
    messages: List[HumanMessage | AIMessage | ToolMessage] = [HumanMessage(content=prompt)]
    last_ai: AIMessage | None = None

    for _ in range(max_steps):
        ai: AIMessage = llm.invoke(messages)
        messages.append(ai)
        last_ai = ai

        calls = getattr(ai, "tool_calls", None) or []
        if not calls:
            # Model produced a final answer
            return messages[-1].content

        # Execute tool calls and feed observations back
        for call in calls:
            name = call["name"]
            args = call.get("args", {}) or {}
            result = tool_map[name].invoke(args)
            messages.append(ToolMessage(
                content=str(result),
                name=name,
                tool_call_id=call.get("id")
            ))

    # If we exit the loop without a clean final AI message, force a wrap up
    messages.append(HumanMessage(content="""
Finish now. Give a short final answer in this exact format:

Current temperature:
Square of current temperature:
""".strip()))
    final_ai: AIMessage = llm.invoke(messages)
    return final_ai.content

print(run_once("""Two step task.

Step 1: Use internet_search to get the current air temperature in New York City today. Show the exact query you used, the top source title and snippet, and extract a numeric temperature in Celsius. Return this temperature as feedback for Step 2.

Step 2: Using the Celsius value from Step 1, compute its square with calculator. Show the exact expression you used and the numeric result.

Important: Give a short final answer in this format:
Current temperature:
Square of current temperature:"""))

Current temperature: 16°C
Square of current temperature: 256

工具

@tool("internet_search")
def internet_search(query: str) -> str:
    """Search Google via SerpAPI for up to date information."""
    serp_api_key = os.environ["SERPAPI_API_KEY"]
    params = {"engine": "google", "gl": "us", "hl": "en"}
    search = SerpAPIWrapper(params=params, serpapi_api_key=serp_api_key)
    return search.run(query)

@tool("calculator")
def calculator(expression: str) -> str:
    """Evaluate a single line mathematical expression with numexpr."""
    local_dict = {"pi": math.pi, "e": math.e}
    out = numexpr.evaluate(
        expression.strip(),
        global_dict={},
        local_dict=local_dict,
    )
    return str(out)

tools = [internet_search, calculator]

用 state、节点和路由构建最小 LangGraph

class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], add_messages]

# LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0, max_tokens=800).bind_tools(tools)

def llm_node(state: AgentState) -> AgentState:
    ai = llm.invoke(state["messages"])
    return {"messages": [ai]}

tool_node = ToolNode(tools=tools)

graph = StateGraph(AgentState)
graph.add_node("llm", llm_node)
graph.add_node("tools", tool_node)
graph.add_edge(START, "llm")

def route(state: AgentState):
    last = state["messages"][-1]
    calls = getattr(last, "tool_calls", None) or []
    return "tools" if calls else END

graph.add_conditional_edges("llm", route, {"tools": "tools", END: END})
graph.add_edge("tools", "llm")

用内存 checkpoint 编译并配置 thread

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

追踪执行并检查 state 与记忆

def _short(msg: BaseMessage, max_len: int = 140) -> str:
    """Compact one-line view of a message."""
    role = type(msg).__name__.replace("Message", "").lower()
    content = getattr(msg, "content", "")
    if isinstance(content, list):
        # some tool outputs can be list payloads
        try:
            content = json.dumps(content)
        except Exception:
            content = str(content)
    text = str(content).replace("\n", " ").strip()
    if len(text) > max_len:
        text = text[: max_len - 3] + "..."
    # include tool name or function call info when available
    if hasattr(msg, "tool_calls") and getattr(msg, "tool_calls"):
        tnames = [tc.get("name", "tool") for tc in msg.tool_calls]
        return f"{role}: tool_calls -> {tnames}"
    if isinstance(msg, ToolMessage):
        return f"{role}({msg.name}): {text}"
    return f"{role}: {text}"

追踪执行并检查 state 与记忆

def print_state_snapshot(app, config, title: str):
    """Print current graph state and memory for a given thread."""
    snap = app.get_state(config)
    values = snap.values or {}
    msgs: List[BaseMessage] = values.get("messages", [])
    print(f"\n=== {title} | state snapshot ===")
    print(f"messages: {len(msgs)} total")
    for i, m in enumerate(msgs[-5:], start=max(0, len(msgs)-5) + 1):
        print(f"  {i:>3}: {_short(m)}")
    # show routing info and queued tasks if present
    nxt = getattr(snap, "next", None)
    tasks = getattr(snap, "tasks", None)
    if nxt:
        print(f"next nodes: {list(nxt)}")
    if tasks:
        print(f"queued tasks: {tasks}")
    # minimal memory view via checkpointer for this thread
    # MemorySaver keeps one latest checkpoint per thread by default, so show existence
    print("memory: in-memory checkpoint present for this thread")

追踪执行并检查 state 与记忆

def run_with_tracing(app, input_state: AgentState, config, title: str):
    """Run the graph while printing per-node updates and final memory."""
    print(f"\n=== {title} | execution trace ===")
    final = None
    # stream_mode="updates" surfaces node-level updates
    for event in app.stream(input_state, config=config, stream_mode="updates"):
        for node, upd in event.items():
            # upd is a dict like {"messages": []} or tool results
            keys = list(upd.keys())
            print(f"[enter {node}] updated: {keys}")
            # if messages updated, print the last one briefly
            msgs = upd.get("messages") or []
            if msgs:
                print(f"  {_short(msgs[-1])}")
            print(f"[leave {node}]")
            final = upd
    # show final assistant message from app.get_state
    print_state_snapshot(app, config, title=f"{title} | after run")
    snap = app.get_state(config)
    msgs = snap.values.get("messages", [])
    return msgs[-1].content if msgs else ""

# configs
cfg = {"configurable": {"thread_id": "nyc-weather-session"}}

第 1 轮：获取纽约市当前摄氏气温

turn1_answer = run_with_tracing(
    app,
    {"messages": [HumanMessage(content="Get the current air temperature in New York City in Celsius.")]},
    config={**cfg, "recursion_limit": 20},
    title="TURN 1",
)
print("\nTURN 1 (final assistant):\n", turn1_answer)

第 2 轮：在同一 thread 中计算该温度的平方

turn2_answer = run_with_tracing(
    app,
    {"messages": [HumanMessage(content="Now compute the square of that temperature.")]},
    config={**cfg, "recursion_limit": 20},
    title="TURN 2",
)
print("\nTURN 2 (final assistant):\n", turn2_answer)

分支一条并行 thread 做不同的追问

cfg_branch = {"configurable": {"thread_id": "nyc-weather-session-branch"}}
branch_answer = run_with_tracing(
    app,
    {"messages": [HumanMessage(content="Instead of squaring, convert it to Fahrenheit and report both.")]},
    config=cfg_branch,
    title="BRANCH",
)
print("\nBRANCH (final assistant):\n", branch_answer)

# show consolidated memory views for both threads
print_state_snapshot(app, cfg, title="MAIN THREAD memory view")
print_state_snapshot(app, cfg_branch, title="BRANCH THREAD memory view")

=== TURN 1 | execution trace ===
[enter llm] updated: ['messages']
  ai: tool_calls -> ['internet_search']
[leave llm]
[enter tools] updated: ['messages']
  tool(internet_search): {'type': 'weather_result', 'temperature': '18', 'unit': 'Celsius', 'precipitation': '0%', 'humidity': '80%', 'wind': '23 km/h', 'location...
[leave tools]
[enter llm] updated: ['messages']
  ai: The current air temperature in New York City is 18°C.
[leave llm]

=== TURN 1 | after run | state snapshot ===
messages: 4 total
    1: human: Get the current air temperature in New York City in Celsius.
    2: ai: tool_calls -> ['internet_search']
    3: tool(internet_search): {'type': 'weather_result', 'temperature': '18', 'unit': 'Celsius', 'precipitation': '0%', 'humidity': '80%', 'wind': '23 km/h', 'location...
    4: ai: The current air temperature in New York City is 18°C.
memory: in-memory checkpoint present for this thread

TURN 1 (final assistant):
 The current air temperature in New York City is 18°C.

=== TURN 2 | execution trace ===
[enter llm] updated: ['messages']
  ai: tool_calls -> ['calculator']
[leave llm]
[enter tools] updated: ['messages']
  tool(calculator): 16
[leave tools]
[enter llm] updated: ['messages']
  ai: The square of the temperature, 18°C, is 324.
[leave llm]

=== TURN 2 | after run | state snapshot ===
messages: 8 total
    4: ai: The current air temperature in New York City is 18°C.
    5: human: Now compute the square of that temperature.
    6: ai: tool_calls -> ['calculator']
    7: tool(calculator): 16
    8: ai: The square of the temperature, 18°C, is 324.
memory: in-memory checkpoint present for this thread

TURN 2 (final assistant):
 The square of the temperature, 18°C, is 324.

=== BRANCH | execution trace ===
[enter llm] updated: ['messages']
  ai: Sure, I can help with that. Please provide the temperature in Celsius that you would like to convert to Fahrenheit.
[leave llm]

=== BRANCH | after run | state snapshot ===
messages: 2 total
    1: human: Instead of squaring, convert it to Fahrenheit and report both.
    2: ai: Sure, I can help with that. Please provide the temperature in Celsius that you would like to convert to Fahrenheit.
memory: in-memory checkpoint present for this thread

BRANCH (final assistant):
 Sure, I can help with that. Please provide the temperature in Celsius that you would like to convert to Fahrenheit.

=== MAIN THREAD memory view | state snapshot ===
messages: 8 total
    4: ai: The current air temperature in New York City is 18°C.
    5: human: Now compute the square of that temperature.
    6: ai: tool_calls -> ['calculator']
    7: tool(calculator): 16
    8: ai: The square of the temperature, 18°C, is 324.
memory: in-memory checkpoint present for this thread

=== BRANCH THREAD memory view | state snapshot ===
messages: 2 total
    1: human: Instead of squaring, convert it to Fahrenheit and report both.
    2: ai: Sure, I can help with that. Please provide the temperature in Celsius that you would like to convert to Fahrenheit.
memory: in-memory checkpoint present for this thread