返回AI笔记
AI Day 55

AI Day 55: 实战(5)：Agent开发实战 — 构建能用工具的AI助手

AI Day 55: 实战(5)：Agent开发实战 — 构建能用工具的AI助手

2026-05-26

日期: 2026-05-26 | 阶段: 第五阶段 · 动手实战 (Day 51-60) | 主题: Agent Development in Practice

学习路径 / Learning Path

AI/LLM 深度技术学习 60天计划
├── 第一阶段：模型基础 (Day 1-15) ✅
│   ├── Day 1: Transformer与LLM基础 ✅
│   ├── Day 2: 量化与本地部署 ✅
│   ├── Day 3: 训练全流程 ✅
│   ├── Day 4: Prompt Engineering ✅
│   ├── Day 5: RAG架构 ✅
│   ├── Day 6: 向量数据库与Embedding ✅
│   ├── Day 7: 微调技术 ✅
│   ├── Day 8: 推理优化 ✅
│   ├── Day 9: 长上下文技术 ✅
│   ├── Day 10: 多模态模型 ✅
│   ├── Day 11: 推理模型 ✅
│   ├── Day 12: Agent框架 ✅
│   ├── Day 13: MCP协议 ✅
│   ├── Day 14: 模型评估 ✅
│   └── Day 15: 阶段一总结 ✅
├── 第二阶段：工程实践 (Day 16-30) ✅
│   ├── Day 16: LLM应用架构 ✅
│   ├── Day 17: 安全与护栏 ✅
│   ├── Day 18: 可观测性 ✅
│   ├── Day 19: 生产RAG·解析与分块 ✅
│   ├── Day 20: 生产RAG·检索与重排 ✅
│   ├── Day 21: 生产RAG·评估与迭代 ✅
│   ├── Day 22: Agent状态与恢复 ✅
│   ├── Day 23: Agent成本优化 ✅
│   ├── Day 24: 多Agent系统 ✅
│   ├── Day 25: Agent测试部署 ✅
│   ├── Day 26: LLM成本工程 ✅
│   ├── Day 27: 多模型编排 ✅
│   ├── Day 28: LLM应用测试 ✅
│   ├── Day 29: 企业LLM平台 ✅
│   └── Day 30: 阶段二总结 ✅
├── 第三阶段：金融零售AI应用 (Day 31-42) ✅
│   ├── Day 31: 金融AI风控 ✅
│   ├── Day 32: 智能投顾与量化 ✅
│   ├── Day 33: 合规与RegTech ✅
│   ├── Day 34: 信贷AI全链路 ✅
│   ├── Day 35: 金融AI总结 ✅
│   ├── Day 36: 零售AI推荐 ✅
│   ├── Day 37: 智能客服 ✅
│   ├── Day 38: 供应链AI ✅
│   ├── Day 39: 智能营销 ✅
│   ├── Day 40: 零售AI总结 ✅
│   ├── Day 41: CeFi×DeFi×AI融合 ✅
│   └── Day 42: AI融合案例与职业 ✅
├── 第四阶段：面试冲刺 (Day 43-50) ✅
│   ├── Day 43: 系统设计·LLM平台 ✅
│   ├── Day 44: 系统设计·RAG系统 ✅
│   ├── Day 45: 系统设计·Agent系统 ✅
│   ├── Day 46: 系统设计·推荐系统 ✅
│   ├── Day 47: 面试·产品AI ✅
│   ├── Day 48: 面试·架构AI ✅
│   ├── Day 49: 面试·行为AI ✅
│   └── Day 50: 学习总结 ✅
└── 第五阶段：动手实战 (Day 51-60)
    ├── Day 51: 本地大模型部署全流程 ✅
    ├── Day 52: RAG系统实战：从文档到问答 ✅
    ├── Day 53: RAG进阶：评估优化与生产化 ✅
    ├── Day 54: LoRA微调实战：训练你的专属模型 ✅
    ├── Day 55: Agent开发实战：构建工具调用Agent ← 你在这里
    ├── Day 56: MCP Server开发：扩展AI能力边界
    ├── Day 57: 多模态应用：图文理解与文档分析
    ├── Day 58: AI应用全栈开发：前后端集成
    ├── Day 59: 性能调优与成本实战
    └── Day 60: 总结与作品集

核心概念 / Core Concepts

Agent = LLM + Tools + Memory / 从"回答问题"到"完成任务"

前几天做的事情 vs 今天做的事情：

Day 52-53 RAG:
  用户提问 → 检索文档 → 生成回答
  LLM 的角色：回答者（被动）
  能力边界：只能用检索到的信息

Day 54 微调:
  训练数据 → 优化模型权重 → 更好地回答
  LLM 的角色：知识库（被动）
  能力边界：只能用训练过的知识

Day 55 Agent:
  用户任务 → 思考 → 使用工具 → 观察结果 → 继续思考 → 完成任务
  LLM 的角色：决策者（主动）
  能力边界：工具有什么能力，Agent就有什么能力

Agent 的本质：
  LLM = 大脑（推理和决策）
  Tools = 手和脚（执行动作）
  Memory = 记忆（保持上下文）

  传统程序: if/else → 执行 (确定性)
  Agent:     思考 → 决策 → 执行 → 反思 (自主性)

Day 12 理论 → Day 55 实践 / Theory to Practice

Day 12 学了什么：
├── Agent 的架构模式（ReAct、Plan-and-Execute）
├── Tool Calling 机制
├── 多 Agent 协作
└── 框架对比（LangChain、AutoGPT、CrewAI）

Day 22-25 学了什么：
├── Agent 状态管理与错误恢复
├── Agent 成本优化
├── 多 Agent 通信模式
└── Agent 测试部署

Day 45 设计了什么：
├── Agent 系统架构（面试题）
└── 安全、监控、回退策略

今天要做什么：
  把上面所有理论，变成一个能运行的 Agent！

框架选择：LangGraph
  Day 12 对比过框架，选 LangGraph 因为：
  1. 状态管理最清晰（Graph + State）
  2. 可视化调试友好
  3. LangSmith 集成
  4. 适合复杂 Agent（多步骤、条件分支）

知识点1：LangGraph 入门 / LangGraph Fundamentals

核心概念 / Core Concepts

LangGraph 的四个基本元素：

1. State（状态）
   Agent 在执行过程中的"记忆"
   包含：消息历史、工具调用结果、中间变量
   每一步都会更新 State

2. Node（节点）
   执行具体操作的函数
   例如：调用 LLM、执行工具、处理结果
   每个节点接收 State，返回更新后的 State

3. Edge（边）
   节点之间的连接关系
   决定执行流程：顺序、条件分支、循环

4. Graph（图）
   由 Node + Edge 组成的完整工作流
   定义了 Agent 的整个执行逻辑

类比：
  State = 你脑中的想法
  Node  = 你做的每个动作
  Edge  = 动作之间的逻辑关系
  Graph = 整个任务的执行计划

安装配置 / Installation

# === Agent 开发依赖 ===
pip install langgraph langchain langchain-community langchain-openai
pip install tavily-python     # Web 搜索工具
pip install httpx             # HTTP 请求
pip install python-dotenv     # 环境变量管理

# === 可选：可视化和调试 ===
pip install langsmith         # Trace 和调试
pip install grandalf          # Graph 可视化

第一个 Agent：计算器 Agent / First Agent: Calculator

"""
calculator_agent.py
最简单的 Agent — 能做数学计算

为什么从计算器开始？
  1. 工具逻辑简单，容易调试
  2. 能清楚展示 Agent 的 ReAct 循环
  3. LLM 本身不擅长数学，工具价值明显
"""

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, SystemMessage
import json
import math

# ==========================================
# 1. 定义状态
# ==========================================

class AgentState(TypedDict):
    """Agent 的状态定义"""
    messages: Annotated[list, add_messages]
    # add_messages 会自动合并新消息到历史中

# ==========================================
# 2. 定义工具
# ==========================================

@tool
def calculator(expression: str) -> str:
    """计算数学表达式。输入一个数学表达式字符串，返回计算结果。
    例如: "2 + 3 * 4" 返回 "14"
    支持: +, -, *, /, **, sqrt, sin, cos, log
    """
    try:
        # 安全的数学运算环境
        allowed_names = {
            "sqrt": math.sqrt,
            "sin": math.sin,
            "cos": math.cos,
            "tan": math.tan,
            "log": math.log,
            "log10": math.log10,
            "pi": math.pi,
            "e": math.e,
            "abs": abs,
            "round": round,
        }
        result = eval(expression, {"__builtins__": {}}, allowed_names)
        return f"计算结果: {expression} = {result}"
    except Exception as e:
        return f"计算错误: {e}"

# 工具列表
tools = [calculator]

# ==========================================
# 3. 创建 LLM (带工具绑定)
# ==========================================

# 使用 Ollama 本地模型
llm = ChatOpenAI(
    base_url="http://localhost:11434/v1",
    api_key="ollama",
    model="qwen2.5:7b",
    temperature=0,
)

# 绑定工具 — 让 LLM 知道有哪些工具可用
llm_with_tools = llm.bind_tools(tools)

# ==========================================
# 4. 定义节点
# ==========================================

def agent_node(state: AgentState) -> AgentState:
    """Agent 思考节点：调用 LLM 决定下一步"""
    system_msg = SystemMessage(
        content="你是一个数学助手。当需要计算时，使用 calculator 工具。"
    )
    messages = [system_msg] + state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

def tool_node(state: AgentState) -> AgentState:
    """工具执行节点：执行 LLM 选择的工具"""
    last_message = state["messages"][-1]
    results = []

    for tool_call in last_message.tool_calls:
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]

        # 找到对应工具并执行
        tool_fn = {t.name: t for t in tools}.get(tool_name)
        if tool_fn:
            result = tool_fn.invoke(tool_args)
            results.append({
                "role": "tool",
                "content": str(result),
                "tool_call_id": tool_call["id"],
            })

    return {"messages": results}

# ==========================================
# 5. 定义边（路由逻辑）
# ==========================================

def should_use_tool(state: AgentState) -> str:
    """决定是否需要调用工具"""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"      # 需要调用工具
    return END              # 直接结束（已有答案）

# ==========================================
# 6. 构建 Graph
# ==========================================

graph = StateGraph(AgentState)

# 添加节点
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)

# 设置入口
graph.set_entry_point("agent")

# 添加边
graph.add_conditional_edges(
    "agent",
    should_use_tool,
    {
        "tools": "tools",   # 如果需要工具 → 执行工具
        END: END,            # 如果不需要 → 结束
    },
)
graph.add_edge("tools", "agent")  # 工具执行后 → 回到 Agent 思考

# 编译
app = graph.compile()

# ==========================================
# 7. 运行测试
# ==========================================

def run_agent(question: str):
    """运行 Agent 并打印每一步"""
    print(f"\n{'='*60}")
    print(f"Question: {question}")
    print(f"{'='*60}")

    result = app.invoke({
        "messages": [HumanMessage(content=question)]
    })

    # 打印完整的消息链
    for i, msg in enumerate(result["messages"]):
        role = msg.__class__.__name__
        content = msg.content[:200] if msg.content else "[tool_call]"
        print(f"\n  Step {i}: [{role}]")
        print(f"  {content}")

        if hasattr(msg, "tool_calls") and msg.tool_calls:
            for tc in msg.tool_calls:
                print(f"  → Tool: {tc['name']}({tc['args']})")

    return result


if __name__ == "__main__":
    # 简单计算
    run_agent("请计算 (15 + 27) * 3 - 10")

    # 多步计算
    run_agent("一个圆的半径是5，求它的面积和周长")

    # 不需要工具的问题
    run_agent("你好，你是谁？")

知识点2：工具开发 / Tool Development

自定义工具集 / Custom Tool Suite

"""
tools.py
Agent 的工具集 — 每个工具扩展 Agent 的一个能力

Day 12 学过：
  "Agent 的能力 = LLM 的推理能力 × 工具的执行能力"
  "工具设计的好坏直接影响 Agent 的表现"

工具设计原则：
1. 单一职责：每个工具只做一件事
2. 清晰描述：LLM 靠描述决定是否使用
3. 输入验证：防止 LLM 传入非法参数
4. 错误处理：优雅地处理失败
"""

from langchain_core.tools import tool
from typing import Optional
import httpx
import json
from pathlib import Path

# ==========================================
# 工具1: 文件读取
# ==========================================

@tool
def read_file(file_path: str) -> str:
    """读取本地文件内容。
    输入文件的绝对路径或相对路径，返回文件的文本内容。
    支持 .md, .txt, .py, .json 等文本文件。
    如果文件不存在或无法读取，返回错误信息。
    """
    try:
        path = Path(file_path)

        # 安全检查：只允许读取特定目录
        allowed_dirs = ["docs/", "src/", "notes/"]
        is_allowed = any(
            str(path).replace("\\", "/").startswith(d)
            for d in allowed_dirs
        )

        if not is_allowed:
            return f"安全限制：不允许读取 {file_path}，只能访问 docs/, src/, notes/ 目录"

        if not path.exists():
            return f"文件不存在: {file_path}"

        content = path.read_text(encoding="utf-8")

        # 截断过长内容
        if len(content) > 3000:
            content = content[:3000] + f"\n\n... (truncated, total {len(content)} chars)"

        return content

    except Exception as e:
        return f"读取文件错误: {e}"


# ==========================================
# 工具2: Web 搜索
# ==========================================

@tool
def web_search(query: str, max_results: int = 3) -> str:
    """搜索互联网获取最新信息。
    输入搜索关键词，返回搜索结果的摘要。
    适用于需要最新数据、新闻、或模型知识库中没有的信息。
    max_results: 返回结果数量，默认3条。
    """
    try:
        # 方案1: 使用 Tavily API (推荐，专为 AI Agent 设计)
        from tavily import TavilyClient
        import os

        api_key = os.getenv("TAVILY_API_KEY")
        if not api_key:
            return "错误: 未配置 TAVILY_API_KEY 环境变量"

        client = TavilyClient(api_key=api_key)
        results = client.search(query=query, max_results=max_results)

        formatted = []
        for i, result in enumerate(results.get("results", []), 1):
            formatted.append(
                f"[{i}] {result['title']}\n"
                f"    URL: {result['url']}\n"
                f"    摘要: {result['content'][:200]}"
            )

        return "\n\n".join(formatted) if formatted else "未找到相关结果"

    except ImportError:
        # 方案2: 简化版 — 直接告诉 Agent 搜索不可用
        return (
            f"Web搜索暂不可用（未安装 tavily-python）。"
            f"请用其他方式获取关于「{query}」的信息。"
        )
    except Exception as e:
        return f"搜索错误: {e}"


# ==========================================
# 工具3: 数据库查询 (SQLite)
# ==========================================

@tool
def query_database(sql: str, db_path: str = "data/notes.db") -> str:
    """执行 SQL 查询获取结构化数据。
    输入 SQL 语句（仅支持 SELECT），返回查询结果。
    数据库包含学习笔记的元数据。
    表结构: notes(id, title, day, category, tags, created_at)
    """
    import sqlite3

    # 安全检查：只允许 SELECT
    sql_upper = sql.strip().upper()
    if not sql_upper.startswith("SELECT"):
        return "安全限制: 只允许 SELECT 查询"

    dangerous_keywords = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "CREATE"]
    if any(kw in sql_upper for kw in dangerous_keywords):
        return f"安全限制: 不允许修改数据库的操作"

    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.execute(sql)
        columns = [desc[0] for desc in cursor.description]
        rows = cursor.fetchall()
        conn.close()

        if not rows:
            return "查询结果为空"

        # 格式化输出
        result = f"列: {', '.join(columns)}\n"
        result += f"行数: {len(rows)}\n\n"
        for row in rows[:20]:  # 最多返回20行
            result += " | ".join(str(v) for v in row) + "\n"

        if len(rows) > 20:
            result += f"\n... (共 {len(rows)} 行，显示前 20 行)"

        return result

    except Exception as e:
        return f"查询错误: {e}"


# ==========================================
# 工具4: API 调用 (通用 HTTP)
# ==========================================

@tool
def call_api(url: str, method: str = "GET", params: Optional[str] = None) -> str:
    """调用外部 API 获取数据。
    url: API 的完整 URL
    method: HTTP 方法，GET 或 POST
    params: JSON 格式的参数字符串（可选）
    返回 API 响应内容。
    """
    try:
        # URL 白名单 (安全考虑)
        allowed_domains = [
            "api.coingecko.com",
            "api.llama.fi",        # DeFiLlama
            "api.etherscan.io",
        ]

        from urllib.parse import urlparse
        domain = urlparse(url).netloc
        if domain not in allowed_domains:
            return f"安全限制: 只允许访问 {', '.join(allowed_domains)}"

        headers = {"Accept": "application/json"}

        if method.upper() == "GET":
            response = httpx.get(url, headers=headers, timeout=10)
        elif method.upper() == "POST":
            body = json.loads(params) if params else {}
            response = httpx.post(url, json=body, headers=headers, timeout=10)
        else:
            return f"不支持的 HTTP 方法: {method}"

        response.raise_for_status()

        # 截断过长响应
        content = response.text
        if len(content) > 3000:
            content = content[:3000] + "... (truncated)"

        return content

    except httpx.TimeoutException:
        return f"API 请求超时: {url}"
    except httpx.HTTPStatusError as e:
        return f"API 返回错误: {e.response.status_code}"
    except Exception as e:
        return f"API 调用错误: {e}"


# ==========================================
# 工具注册表
# ==========================================

ALL_TOOLS = [
    calculator,     # 从知识点1导入
    read_file,
    web_search,
    query_database,
    call_api,
]

def get_tools(names: list[str] = None) -> list:
    """获取工具列表，可按名称过滤"""
    if names is None:
        return ALL_TOOLS
    return [t for t in ALL_TOOLS if t.name in names]

工具装饰器详解 / Tool Decorator Deep Dive

@tool 装饰器做了什么？

1. 自动从 docstring 生成工具描述
   LLM 用这个描述决定是否调用工具

2. 自动从类型注解生成参数 schema
   LLM 知道应该传什么参数

3. 包装成标准的 Tool 对象
   框架能统一管理和调用

关键原则：docstring 要写给 LLM 看

好的描述:
  "搜索互联网获取最新信息。
   输入搜索关键词，返回搜索结果的摘要。
   适用于需要最新数据、新闻、或模型知识库中没有的信息。"

差的描述:
  "Web search function."
  → LLM 不知道什么时候该用它

LLM 的决策过程：
  1. 看到用户问题
  2. 看所有工具的描述
  3. 选择最匹配的工具
  4. 根据参数 schema 构造调用参数

所以：工具描述 = 给 LLM 的使用说明书

知识点3：ReAct Agent 实现 / ReAct Agent Implementation

完整的 ReAct 循环 / Full ReAct Loop

"""
react_agent.py
完整的 ReAct Agent — Thought → Action → Observation → Answer

Day 12 学的 ReAct 模式：
  Reasoning + Acting 交替进行
  让 LLM 先"想"再"做"，做完"看"结果，再继续"想"

今天实现完整版本，支持：
  - 多轮工具调用
  - 错误重试
  - 最大步数限制（防无限循环）
  - 详细日志
"""

from typing import Annotated, TypedDict, Literal
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from langchain_core.messages import (
    HumanMessage, SystemMessage, AIMessage, ToolMessage,
)
import time

# ==========================================
# 1. 增强的 State 定义
# ==========================================

class ReActState(TypedDict):
    """ReAct Agent 状态"""
    messages: Annotated[list, add_messages]
    step_count: int        # 当前步数
    max_steps: int         # 最大步数限制

# ==========================================
# 2. 系统提示词
# ==========================================

REACT_SYSTEM_PROMPT = """你是一个智能助手，能够使用工具来完成任务。

## 工作方式
1. 先分析用户的请求，理解需要做什么
2. 如果需要外部信息或计算，使用合适的工具
3. 根据工具返回的结果，组织回答
4. 如果一次工具调用不够，可以连续使用多个工具

## 注意事项
- 优先使用工具获取准确数据，不要猜测
- 如果工具调用失败，尝试换个方式
- 每一步都要解释你的思考过程
- 最终回答要全面、有条理

## 可用工具
你可以使用以下工具（具体描述见工具定义）：
- calculator: 数学计算
- read_file: 读取本地文件
- web_search: 搜索互联网
- query_database: 查询数据库
- call_api: 调用外部API
"""

# ==========================================
# 3. Agent 节点
# ==========================================

from tools import ALL_TOOLS

llm = ChatOpenAI(
    base_url="http://localhost:11434/v1",
    api_key="ollama",
    model="qwen2.5:7b",
    temperature=0,
)

llm_with_tools = llm.bind_tools(ALL_TOOLS)

def agent_node(state: ReActState) -> dict:
    """Agent 思考和决策节点"""
    step = state.get("step_count", 0)
    max_steps = state.get("max_steps", 10)

    # 步数限制保护
    if step >= max_steps:
        return {
            "messages": [
                AIMessage(content=f"已达到最大步数限制({max_steps})，基于已有信息给出回答。")
            ],
            "step_count": step + 1,
        }

    # 构造消息
    system_msg = SystemMessage(content=REACT_SYSTEM_PROMPT)
    messages = [system_msg] + state["messages"]

    # 调用 LLM
    start = time.time()
    response = llm_with_tools.invoke(messages)
    elapsed = time.time() - start

    # 日志
    print(f"\n  [Step {step + 1}] Agent thinking... ({elapsed:.1f}s)")
    if response.content:
        print(f"  Thought: {response.content[:150]}...")
    if hasattr(response, "tool_calls") and response.tool_calls:
        for tc in response.tool_calls:
            print(f"  Action: {tc['name']}({json.dumps(tc['args'], ensure_ascii=False)[:100]})")

    return {
        "messages": [response],
        "step_count": step + 1,
    }

# ==========================================
# 4. 工具执行节点 (使用 LangGraph 内置)
# ==========================================

tool_node = ToolNode(ALL_TOOLS)

def tool_executor(state: ReActState) -> dict:
    """工具执行节点 — 包装 ToolNode，添加日志"""
    last_message = state["messages"][-1]

    print(f"\n  [Tool Execution]")
    start = time.time()

    # 执行工具
    result = tool_node.invoke(state)

    elapsed = time.time() - start

    # 打印工具结果
    for msg in result.get("messages", []):
        if hasattr(msg, "content"):
            print(f"  Observation: {str(msg.content)[:200]}...")

    print(f"  Tool completed in {elapsed:.1f}s")

    return result

# ==========================================
# 5. 路由逻辑
# ==========================================

import json

def route_agent(state: ReActState) -> Literal["tools", "__end__"]:
    """决定下一步：调用工具还是结束"""
    last_message = state["messages"][-1]
    step = state.get("step_count", 0)
    max_steps = state.get("max_steps", 10)

    # 达到步数限制 → 结束
    if step >= max_steps:
        return END

    # 有工具调用 → 执行工具
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"

    # 没有工具调用 → Agent 已给出最终回答
    return END

# ==========================================
# 6. 构建 Graph
# ==========================================

graph = StateGraph(ReActState)

graph.add_node("agent", agent_node)
graph.add_node("tools", tool_executor)

graph.set_entry_point("agent")

graph.add_conditional_edges("agent", route_agent)
graph.add_edge("tools", "agent")  # 工具完成 → 回到 Agent

react_agent = graph.compile()

# ==========================================
# 7. 运行示例
# ==========================================

def run_react(question: str, max_steps: int = 10):
    """运行 ReAct Agent"""
    print(f"\n{'='*60}")
    print(f"Task: {question}")
    print(f"Max Steps: {max_steps}")
    print(f"{'='*60}")

    result = react_agent.invoke({
        "messages": [HumanMessage(content=question)],
        "step_count": 0,
        "max_steps": max_steps,
    })

    # 提取最终回答
    final = result["messages"][-1]
    print(f"\n{'='*60}")
    print(f"Final Answer:")
    print(f"{final.content}")
    print(f"Total Steps: {result['step_count']}")
    print(f"{'='*60}")

    return result


if __name__ == "__main__":
    # 示例1: 需要计算的问题
    run_react("一个圆柱体底面半径3cm，高10cm，求体积和表面积")

    # 示例2: 需要文件读取的问题
    run_react("读取 docs/ai/day7-finetuning-lora-qlora.md 的内容，总结LoRA的核心原理")

    # 示例3: 多工具组合
    run_react("查看Day 12的Agent框架笔记，总结ReAct模式，然后计算如果每个Agent步骤耗时2秒，10步Agent完成一个任务需要多少秒")

实际日志展示 / Real Execution Logs

============================================================
Task: 一个圆柱体底面半径3cm，高10cm，求体积和表面积
Max Steps: 10
============================================================

  [Step 1] Agent thinking... (2.3s)
  Thought: 需要计算圆柱体的体积和表面积。
    体积 = π * r² * h
    表面积 = 2πr(r + h)
    我来用计算器工具分别算。
  Action: calculator({"expression": "3.14159 * 3**2 * 10"})

  [Tool Execution]
  Observation: 计算结果: 3.14159 * 3**2 * 10 = 282.7431
  Tool completed in 0.0s

  [Step 2] Agent thinking... (1.8s)
  Thought: 体积计算完成，约282.74 cm³。现在计算表面积。
  Action: calculator({"expression": "2 * 3.14159 * 3 * (3 + 10)"})

  [Tool Execution]
  Observation: 计算结果: 2 * 3.14159 * 3 * (3 + 10) = 245.0441...
  Tool completed in 0.0s

  [Step 3] Agent thinking... (1.5s)
  Thought: 两个计算都完成了，给出最终答案。

============================================================
Final Answer:
圆柱体的计算结果如下：

**体积** = πr²h = π × 3² × 10 ≈ 282.74 cm³
**表面积** = 2πr(r + h) = 2π × 3 × (3 + 10) ≈ 245.04 cm²

其中 r = 3cm, h = 10cm。
Total Steps: 3
============================================================

Agent 的 ReAct 过程：
  Step 1: Thought(分析问题) → Action(计算体积) → Observation(282.74)
  Step 2: Thought(体积OK) → Action(计算面积) → Observation(245.04)
  Step 3: Thought(整合结果) → Answer(最终回答)

知识点4：记忆系统 / Memory System

短期记忆：对话历史 / Short-term Memory: Conversation History

"""
memory.py
Agent 的记忆系统

Day 22 学过 Agent 状态管理：
  "无状态 Agent 像金鱼，每次对话都从头开始"
  "有记忆的 Agent 像人，能记住上下文"

两种记忆：
  短期记忆 = 当前对话的历史（存在 State 里）
  长期记忆 = 跨对话的知识（存在向量数据库里）
"""

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

class ConversationMemory:
    """对话记忆管理器"""

    def __init__(self, max_turns: int = 10):
        self.max_turns = max_turns
        self.history: list = []

    def add_turn(self, user_msg: str, agent_msg: str):
        """添加一轮对话"""
        self.history.append({
            "user": user_msg,
            "agent": agent_msg,
        })

        # 裁剪过长的历史
        if len(self.history) > self.max_turns:
            self.history = self.history[-self.max_turns:]

    def get_messages(self) -> list:
        """获取消息列表格式的历史"""
        messages = []
        for turn in self.history:
            messages.append(HumanMessage(content=turn["user"]))
            messages.append(AIMessage(content=turn["agent"]))
        return messages

    def get_summary(self) -> str:
        """获取对话摘要（当历史太长时）"""
        if len(self.history) <= 3:
            return ""

        summary_parts = []
        for turn in self.history[:-3]:  # 总结较早的对话
            summary_parts.append(f"- 用户问: {turn['user'][:50]}...")
            summary_parts.append(f"  助手答: {turn['agent'][:50]}...")

        return "之前的对话摘要:\n" + "\n".join(summary_parts)

    def clear(self):
        """清空记忆"""
        self.history = []

长期记忆：向量存储 / Long-term Memory: Vector Store

"""
long_term_memory.py
长期记忆 — 跨对话的知识存储

结合 Day 52 的 ChromaDB 向量数据库
Agent 可以"回忆"之前对话中的重要信息
"""

import chromadb
from datetime import datetime
import json

class LongTermMemory:
    """基于向量数据库的长期记忆"""

    def __init__(self, persist_dir: str = "./agent_memory"):
        self.client = chromadb.PersistentClient(path=persist_dir)
        self.collection = self.client.get_or_create_collection(
            name="agent_memories",
            metadata={"hnsw:space": "cosine"},
        )

    def store(self, content: str, metadata: dict = None):
        """存储一条记忆"""
        memory_id = f"mem_{datetime.now().strftime('%Y%m%d_%H%M%S_%f')}"
        meta = metadata or {}
        meta["timestamp"] = datetime.now().isoformat()

        self.collection.add(
            documents=[content],
            metadatas=[meta],
            ids=[memory_id],
        )

    def recall(self, query: str, n_results: int = 3) -> list[str]:
        """回忆与查询相关的记忆"""
        results = self.collection.query(
            query_texts=[query],
            n_results=n_results,
        )

        memories = []
        for doc, meta in zip(
            results["documents"][0],
            results["metadatas"][0],
        ):
            timestamp = meta.get("timestamp", "unknown")
            memories.append(f"[{timestamp}] {doc}")

        return memories

    def store_conversation_summary(self, user_msg: str, agent_msg: str):
        """存储对话摘要作为长期记忆"""
        summary = f"用户问: {user_msg}\n助手答: {agent_msg[:200]}"
        self.store(
            content=summary,
            metadata={
                "type": "conversation",
                "user_query": user_msg[:100],
            },
        )

    def get_relevant_context(self, question: str) -> str:
        """获取与当前问题相关的历史记忆"""
        memories = self.recall(question, n_results=3)
        if not memories:
            return ""

        return (
            "## 相关历史记忆\n"
            + "\n".join(f"- {m}" for m in memories)
        )

记忆检索与注入 / Memory Retrieval & Injection

"""
memory_agent.py
带记忆的 Agent — 整合短期和长期记忆

Agent 在每次思考前：
1. 从短期记忆获取对话上下文
2. 从长期记忆检索相关知识
3. 将两者注入到 System Prompt 中
"""

class MemoryAgent:
    """带完整记忆系统的 Agent"""

    def __init__(self):
        self.short_memory = ConversationMemory(max_turns=10)
        self.long_memory = LongTermMemory()

    def chat(self, user_input: str) -> str:
        """带记忆的对话"""

        # 1. 检索长期记忆
        relevant_memories = self.long_memory.get_relevant_context(user_input)

        # 2. 构建增强的 System Prompt
        enhanced_prompt = REACT_SYSTEM_PROMPT
        if relevant_memories:
            enhanced_prompt += f"\n\n{relevant_memories}"

        # 3. 获取对话历史
        history_messages = self.short_memory.get_messages()

        # 4. 运行 Agent
        messages = (
            [SystemMessage(content=enhanced_prompt)]
            + history_messages
            + [HumanMessage(content=user_input)]
        )

        result = react_agent.invoke({
            "messages": messages,
            "step_count": 0,
            "max_steps": 10,
        })

        # 5. 提取回答
        answer = result["messages"][-1].content

        # 6. 更新记忆
        self.short_memory.add_turn(user_input, answer)
        self.long_memory.store_conversation_summary(user_input, answer)

        return answer


# 使用示例
agent = MemoryAgent()

# 第一轮对话
agent.chat("LoRA的rank参数一般设置多少？")

# 第二轮对话（Agent 能记住上下文）
agent.chat("那alpha呢？")  # Agent 知道你在问 LoRA 的 alpha

# 下次启动（长期记忆仍在）
agent2 = MemoryAgent()
agent2.chat("我们之前讨论过LoRA参数，能回忆一下吗？")
# → Agent 从长期记忆中检索到之前的对话

知识点5：实战项目 — Web3 研究 Agent / Web3 Research Agent

项目架构 / Project Architecture

Web3 Research Agent 功能：
  输入: 协议名称（如 "Uniswap"）
  过程: 自动收集数据 → 分析 → 生成报告
  输出: 结构化的协议分析报告

数据来源:
├── CoinGecko API: 价格、市值、交易量
├── DeFiLlama API: TVL、协议排名
├── Web 搜索: 最新新闻和动态
└── 本地笔记: 已有的分析和知识

Agent 工作流:
  1. 接收协议名称
  2. 查询 CoinGecko 获取基本数据
  3. 查询 DeFiLlama 获取 TVL 数据
  4. 搜索最新新闻
  5. 综合所有信息生成分析报告

完整代码 / Full Implementation

"""
web3_research_agent.py
Web3 协议研究 Agent

结合了:
  Day 12 的 Agent 架构
  Day 34 的 Web3 数据指标
  Day 52 的 RAG 系统
  → 自动化的协议研究
"""

from langchain_core.tools import tool
import httpx
import json

# ==========================================
# Web3 专用工具
# ==========================================

@tool
def get_token_price(token_id: str) -> str:
    """获取加密货币的当前价格和市场数据。
    token_id: CoinGecko 的代币ID，如 'ethereum', 'uniswap', 'aave'
    返回: 价格、24h变化、市值、24h交易量等数据
    """
    try:
        url = f"https://api.coingecko.com/api/v3/coins/{token_id}"
        params = {
            "localization": "false",
            "tickers": "false",
            "community_data": "false",
            "developer_data": "false",
        }

        response = httpx.get(url, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()

        market = data.get("market_data", {})
        result = {
            "name": data.get("name"),
            "symbol": data.get("symbol", "").upper(),
            "price_usd": market.get("current_price", {}).get("usd"),
            "price_change_24h": market.get("price_change_percentage_24h"),
            "market_cap_usd": market.get("market_cap", {}).get("usd"),
            "volume_24h_usd": market.get("total_volume", {}).get("usd"),
            "ath_usd": market.get("ath", {}).get("usd"),
            "ath_change_pct": market.get("ath_change_percentage", {}).get("usd"),
        }

        return json.dumps(result, indent=2, ensure_ascii=False)

    except httpx.HTTPStatusError as e:
        return f"API错误: {e.response.status_code}. 请检查 token_id 是否正确。"
    except Exception as e:
        return f"获取价格数据失败: {e}"


@tool
def get_protocol_tvl(protocol_slug: str) -> str:
    """获取 DeFi 协议的 TVL（总锁仓量）数据。
    protocol_slug: DeFiLlama 的协议名称，如 'uniswap', 'aave', 'lido'
    返回: 当前TVL、TVL变化、链分布等数据
    """
    try:
        url = f"https://api.llama.fi/protocol/{protocol_slug}"
        response = httpx.get(url, timeout=10)
        response.raise_for_status()
        data = response.json()

        # 提取关键信息
        current_tvl = data.get("currentChainTvls", {})
        total_tvl = sum(
            v for k, v in current_tvl.items()
            if not k.endswith("-staking") and not k.endswith("-borrowed")
        )

        result = {
            "name": data.get("name"),
            "category": data.get("category"),
            "total_tvl_usd": total_tvl,
            "chains": list(current_tvl.keys())[:10],
            "chain_tvls": {
                k: v for k, v in sorted(
                    current_tvl.items(), key=lambda x: x[1], reverse=True
                )[:5]
            },
            "description": data.get("description", "")[:200],
        }

        return json.dumps(result, indent=2, ensure_ascii=False)

    except Exception as e:
        return f"获取TVL数据失败: {e}"


@tool
def search_web3_news(query: str) -> str:
    """搜索 Web3/加密货币相关的最新新闻和动态。
    query: 搜索关键词，如 'Uniswap v4 latest news'
    返回最新的相关新闻摘要。
    """
    # 复用 web_search，添加 Web3 上下文
    enhanced_query = f"{query} crypto DeFi 2026"
    return web_search.invoke({"query": enhanced_query, "max_results": 3})


# ==========================================
# 研究 Agent
# ==========================================

WEB3_RESEARCH_PROMPT = """你是一个专业的 Web3 协议研究分析师。

## 任务
收集并分析指定协议的数据，生成结构化的研究报告。

## 研究流程
1. 先用 get_token_price 获取代币价格和市场数据
2. 再用 get_protocol_tvl 获取 TVL 数据
3. 用 search_web3_news 搜索最新动态
4. 综合所有数据，生成分析报告

## 报告格式
请按以下格式输出研究报告：

### 协议概览
- 名称、类别、核心功能

### 市场数据
- 代币价格、市值、交易量、ATH对比

### TVL分析
- 当前TVL、链分布、趋势判断

### 最新动态
- 近期重要新闻和事件

### 综合评估
- 优势、风险、PM视角的产品洞察

注意：所有数据必须来自工具调用结果，不要编造数据。
"""

# 工具列表
web3_tools = [get_token_price, get_protocol_tvl, search_web3_news]

# 构建 Agent (复用 ReAct 架构)
from langchain_openai import ChatOpenAI

research_llm = ChatOpenAI(
    base_url="http://localhost:11434/v1",
    api_key="ollama",
    model="qwen2.5:7b",
    temperature=0.3,
)

research_llm_with_tools = research_llm.bind_tools(web3_tools)

# ... (使用与知识点3相同的 Graph 架构，替换工具和 Prompt)

def research_protocol(protocol_name: str) -> str:
    """运行协议研究"""
    question = f"请对 {protocol_name} 协议进行全面研究分析。"
    result = run_react(question, max_steps=8)
    return result["messages"][-1].content


# 运行示例
if __name__ == "__main__":
    report = research_protocol("Uniswap")
    print(report)

    # 保存报告
    with open(f"research_uniswap.md", "w", encoding="utf-8") as f:
        f.write(report)

输出示例 / Sample Output

# Uniswap 协议研究报告

## 协议概览
- **名称**: Uniswap
- **类别**: DEX (去中心化交易所)
- **核心功能**: 基于 AMM 的代币交换、流动性提供

## 市场数据
- **代币**: UNI
- **当前价格**: $12.34
- **24h 变化**: +3.2%
- **市值**: $74.04 亿
- **24h 交易量**: $2.89 亿
- **ATH**: $44.97 (当前距ATH -72.6%)

## TVL 分析
- **总 TVL**: $48.2 亿
- **链分布**:
  - Ethereum: $32.1 亿 (66.6%)
  - Arbitrum: $8.5 亿 (17.6%)
  - Polygon: $3.2 亿 (6.6%)
  - Base: $2.8 亿 (5.8%)

## 最新动态
- Uniswap V4 hooks 生态持续扩展
- ...

## 综合评估
- **优势**: DEX 市场份额领先，多链部署完善
- **风险**: 监管不确定性，竞争加剧
- **PM 洞察**: V4 的 hooks 架构开启了第三方创新...

知识点6：Agent 调试 / Agent Debugging

LangSmith Trace 查看 / LangSmith Tracing

"""
debug_setup.py
Agent 调试配置

Day 18 学过可观测性：
  "没有 Trace 的 Agent = 黑盒"
  "出了问题不知道哪一步出错"

LangSmith 提供：
  - 完整的执行链路
  - 每一步的输入输出
  - Token 消耗统计
  - 延迟分析
"""

import os

# 配置 LangSmith (免费版足够)
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGCHAIN_PROJECT"] = "web3-research-agent"

# 配置后，所有 LangChain/LangGraph 调用都会自动上报 Trace
# 在 https://smith.langchain.com 可以查看

"""
Trace 中你能看到：

Run: research_protocol("Uniswap")
├── agent_node (Step 1)
│   ├── Input: [SystemMessage, HumanMessage]
│   ├── LLM Call: 2.3s, 450 tokens
│   └── Output: AIMessage(tool_calls=[get_token_price])
│
├── tool_executor (Step 1)
│   ├── Tool: get_token_price("uniswap")
│   ├── HTTP Request: api.coingecko.com (0.8s)
│   └── Output: ToolMessage(price data)
│
├── agent_node (Step 2)
│   ├── Input: [..., ToolMessage]
│   ├── LLM Call: 1.9s, 380 tokens
│   └── Output: AIMessage(tool_calls=[get_protocol_tvl])
│
├── tool_executor (Step 2)
│   ├── Tool: get_protocol_tvl("uniswap")
│   ├── HTTP Request: api.llama.fi (1.2s)
│   └── Output: ToolMessage(TVL data)
│
└── agent_node (Step 3)
    ├── Input: [..., ToolMessage, ToolMessage]
    ├── LLM Call: 3.1s, 820 tokens
    └── Output: AIMessage(final report)

Total: 3 steps, 9.3s, 1650 tokens
"""

常见问题与解决方案 / Common Issues & Solutions

问题1: 工具调用失败
─────────────────────────────
症状: Agent 调用工具后收到错误信息
原因: API 超时、参数错误、网络问题

解决方案:
  1. 在工具中添加重试逻辑
  2. 返回友好的错误信息（让 Agent 能理解并换个方式）
  3. 设置合理的超时时间

  # 工具中的重试模式
  @tool
  def robust_api_call(url: str) -> str:
      """带重试的 API 调用"""
      for attempt in range(3):
          try:
              response = httpx.get(url, timeout=10)
              return response.text
          except httpx.TimeoutException:
              if attempt < 2:
                  continue
              return "API 超时，请稍后重试或换个数据源"


问题2: 无限循环
─────────────────────────────
症状: Agent 反复调用同一个工具，不给出最终答案
原因: LLM 不知道何时停止、工具返回不够明确

解决方案:
  1. max_steps 限制（已实现）
  2. 在 System Prompt 中明确说"收集够数据后直接给出答案"
  3. 检测重复工具调用并强制结束

  def route_agent(state):
      # 检测重复调用
      recent_tools = []
      for msg in state["messages"][-6:]:
          if hasattr(msg, "tool_calls") and msg.tool_calls:
              recent_tools.extend(tc["name"] for tc in msg.tool_calls)

      # 连续3次调用同一工具 → 强制结束
      if len(recent_tools) >= 3 and len(set(recent_tools[-3:])) == 1:
          return END

      # 正常路由逻辑...


问题3: 偏离主题
─────────────────────────────
症状: Agent 开始做与任务无关的事情
原因: 工具返回了干扰信息、System Prompt 不够明确

解决方案:
  1. 在 System Prompt 中明确任务边界
  2. 工具返回结果时过滤无关内容
  3. 添加"任务完成检查"节点

  # 添加任务完成检查
  def check_completion(state):
      """检查 Agent 是否在正确轨道上"""
      original_task = state["messages"][0].content
      latest_thought = state["messages"][-1].content

      # 简单的相关性检查
      # 生产环境可以用 LLM 判断
      return "continue"  # or "redirect"


问题4: 工具选择错误
─────────────────────────────
症状: Agent 用错了工具（该搜索的时候去算数）
原因: 工具描述不够清晰

解决方案:
  1. 改进工具的 docstring 描述
  2. 在描述中加入"使用场景"和"不适用场景"
  3. 添加"什么时候用这个工具"的示例

  @tool
  def web_search(query: str) -> str:
      """搜索互联网获取最新信息。

      适用场景:
      - 需要最新新闻和动态
      - 需要实时数据（价格、事件）
      - 模型知识库中没有的信息

      不适用场景:
      - 数学计算（用 calculator）
      - 读取本地文件（用 read_file）
      """

今日思考 / Today's Reflections

思考1：Agent 的"智能"来自工具，不是 LLM / Agent's Intelligence Comes from Tools

一个反直觉的发现：

没有工具的 LLM (Day 52-54):
  "请查一下 ETH 现在的价格"
  → "很抱歉，我的数据截止到2024年..."
  → 智能感: ★★

有工具的 Agent (今天):
  "请查一下 ETH 现在的价格"
  → 调用 get_token_price("ethereum")
  → "ETH 当前价格 $3,456.78，24h +2.3%"
  → 智能感: ★★★★★

同样的 LLM，差异在于：
  Agent 能"动手做事"，而不仅仅是"说话"

PM 视角的洞察：
  用户不在乎你用了什么模型
  用户在乎你能帮他做什么
  Agent 的产品价值 = 工具的实用价值 × LLM 的调度能力

所以工具设计是 Agent 产品最重要的工作！

思考2：错误处理是 Agent 的生命线 / Error Handling is Agent's Lifeline

今天调试中遇到最多的问题不是 LLM 不聪明
而是工具出错后 Agent 不知道怎么办

典型场景：
  1. CoinGecko API 超时 → Agent 陷入循环重试
  2. 工具返回空数据 → Agent 开始编造数据
  3. 参数格式错误 → Agent 卡在同一步

Day 22 学的错误恢复策略今天全用上了：
  - 重试 + 指数退避
  - 回退到更简单的方案
  - 向用户报告无法完成

核心教训：
  "Agent 80% 的代码在处理异常情况"
  这和传统软件开发一模一样
  Happy Path 容易，Edge Case 才见真功夫

思考3：Agent 是 PM 的新产品形态 / Agents are a New Product Paradigm

传统产品：
  用户操作 UI → 系统处理 → 返回结果
  PM 设计：页面流程、交互逻辑、数据展示

Agent 产品：
  用户说需求 → Agent 规划 → 使用工具 → 返回结果
  PM 设计：工具能力、Agent 人设、安全边界、成本控制

Agent PM 需要思考的新问题：
  1. 工具选择：给 Agent 什么工具？
     工具太多 → Agent 选择困难
     工具太少 → Agent 能力有限

  2. 安全边界：Agent 能做什么？不能做什么？
     Day 17 学的 Guard Rails 在 Agent 场景更关键

  3. 成本控制：每个请求调多少次工具？
     Day 23 的 Agent 成本优化不是学术话题

  4. 用户信任：Agent 犯错怎么办？
     不像传统产品有确定性的结果
     需要透明度（展示思考过程）和可回退性

这是一个全新的产品设计范式
传统 PM 经验有用但不够
需要理解 AI 的能力和限制

学习资源 / Resources

LangGraph

Agent 开发

调试工具

Web3 API

明日预告 / Tomorrow's Preview

Day 56: MCP Server 开发 — 扩展AI的能力边界

从 Agent 到 MCP：
  今天: 工具直接写在 Agent 代码里
  明天: 工具封装成 MCP Server，任何 AI 客户端都能调用

MCP 的价值：
  Agent 的工具 = 只有你的 Agent 能用
  MCP Server = 任何 AI 应用都能用

  今天写的 Web3 工具 → 明天封装成 MCP Server
  → Claude Code 能直接调用
  → Cursor 能直接调用
  → 任何 MCP 兼容的客户端都能调用

Day 13 学的 MCP 协议理论 → Day 56 的实际开发

明天将实现：
1. MCP SDK 搭建
2. 笔记搜索 MCP Server
3. Web3 数据查询 MCP Server
4. 与 Claude Code 集成测试
5. npm 发布流程

一次开发，处处可用！

Day 55 完成！ 从计算器 Agent 到 Web3 研究 Agent，亲手构建了能使用工具的 AI 助手。 Agent = LLM + Tools + Memory，三者缺一不可。 明天用 MCP 把这些工具开放给整个 AI 生态！

上一篇
Day 54
下一篇
Day 56