8.2 实现核心功能

🎯 小白理解指南：这一节讲什么？

前面是"画蓝图"，这一节是**"盖房子"**——把设计变成可运行的代码！

下面这段代码虽然有点长，但每一部分都对应你学过的知识：

代码部分	对应 Module	作用
AgentState	Module 2 数据结构	定义 Agent 的记忆
@tool	Module 1 函数	定义 Agent 的技能
create_react_agent	Module 3 面向对象	组装 Agent
SqliteSaver	Module 4 文件处理	保存对话历史
logging	Module 5 异常处理	记录运行日志
async/await	Module 6 异步编程	异步执行

别被代码长度吓到，一步一步看注释就懂了！

完整的 Agent 实现

"""
完整的 LangGraph Agent 实现
整合所有 Module 的知识

🎯 小白导读：这段代码做了什么？
1. 定义 Agent 要记住什么（State）
2. 定义 Agent 会什么技能（Tools）
3. 组装成一个完整的 Agent
4. 运行并处理用户查询
"""

from typing import TypedDict, List, Annotated
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain.tools import tool
import asyncio
import logging

# ==================== 第 1 步：配置日志 ====================
# 🎯 日志就像"行车记录仪"，记录程序运行过程，出问题时方便排查
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


# ==================== 第 2 步：定义状态 (Module 2) ====================
# 🎯 AgentState 是 Agent 的"记忆结构"
class AgentState(TypedDict):
    messages: List[BaseMessage]  # 聊天记录
    context: dict                 # 上下文信息
    iteration: int                # 迭代次数（防止死循环）


# ==================== 第 3 步：定义工具 (Module 1, 3) ====================
# 🎯 @tool 装饰器把普通函数变成 Agent 可调用的"技能"

@tool
def search_web(query: str) -> str:
    """搜索网络

    🎯 这是 Agent 的"搜索技能"
    真实项目中，这里会调用 Google/Bing API
    """
    logger.info(f"搜索: {query}")  # 记录日志
    return f"搜索结果: {query}"


@tool
def calculator(expression: str) -> str:
    """计算数学表达式

    🎯 这是 Agent 的"计算技能"
    用 try/except 处理错误，防止崩溃
    """
    try:
        result = eval(expression)  # 计算表达式
        return f"结果: {result}"
    except Exception as e:
        return f"错误: {e}"  # 出错返回错误信息，不崩溃


# ==================== 第 4 步：组装 Agent ====================
# 🎯 把大脑（LLM）+ 技能（Tools）+ 记忆（Memory）组装在一起

def create_production_agent():
    """创建生产级 Agent"""

    # 1. 创建大脑 —— 选择 GPT-4，temperature=0 让回答更稳定
    llm = ChatOpenAI(model="gpt-4", temperature=0)

    # 2. 准备技能 —— 把工具放进列表
    tools = [search_web, calculator]

    # 3. 创建记忆 —— 用 SQLite 保存对话历史
    # 🎯 ":memory:" 表示保存在内存中，程序关闭就清空
    # 真实项目用 "agent.db" 这样的文件名，永久保存
    memory = SqliteSaver.from_conn_string(":memory:")

    # 4. 一键组装！
    # 🎯 create_react_agent 是 LangGraph 的"傻瓜式"创建函数
    agent = create_react_agent(
        model=llm,           # 大脑
        tools=tools,         # 技能
        checkpointer=memory  # 记忆
    )

    return agent


# ==================== 第 5 步：异步运行 (Module 6) ====================
# 🎯 用 async/await 让 Agent 可以同时处理多个请求

async def run_agent_async(query: str):
    """异步运行 Agent"""
    agent = create_production_agent()

    # thread_id 区分不同的对话，就像微信里不同的聊天窗口
    config = {"configurable": {"thread_id": "session_001"}}

    # ainvoke = async invoke，异步调用
    result = await agent.ainvoke(
        {"messages": [HumanMessage(content=query)]},  # 用户的消息
        config=config
    )

    return result


# ==================== 第 6 步：主函数 ====================
# 🎯 程序入口，演示如何使用这个 Agent

def main():
    """主函数"""
    # 准备几个测试问题
    queries = [
        "搜索 Python 教程",
        "计算 123 * 456",
    ]

    # 逐个处理
    for query in queries:
        print(f"\n查询: {query}")
        # asyncio.run() 启动异步程序
        result = asyncio.run(run_agent_async(query))
        # 取最后一条消息（就是 Agent 的回答）
        print(f"结果: {result['messages'][-1].content}")


# 🎯 Python 经典入口：只有直接运行这个文件时才执行 main()
if __name__ == "__main__":
    main()

🎯 小白总结：这段代码的执行流程

main() 启动
  ↓
创建 Agent（大脑 + 技能 + 记忆）
  ↓
用户提问："搜索 Python 教程"
  ↓
Agent 思考："这需要用搜索技能"
  ↓
调用 search_web("Python 教程")
  ↓
返回结果给用户

---

下一节：8.3 测试与部署