Day 15：LangChain 入门

今日目标

用 LangChain 框架重写昨天的 ReAct Agent，体会：

• 框架如何简化代码
• 昨天手写的每个部分对应框架的什么组件
• 框架的优缺点

---

核心概念对照

---

环境准备

pip install langchain langchain-openai

---

第一部分：最简单的 LangChain Agent

from langchain_openai import ChatOpenAI
from langchain.agents import tool, AgentExecutor, create_react_agent
from langchain import hub
import datetime

# 1. 创建 LLM
llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key="你的API Key",
    temperature=0
)

# 2. 定义工具（对比昨天的 @registry.register）
@tool
def get_time() -> str:
    """获取当前时间"""
    return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气
    
    Args:
        city: 城市名称，如北京、上海
    """
    weather_data = {
        "北京": "晴，气温 18-26°C，空气质量良",
        "上海": "多云，气温 20-28°C，空气质量优",
        "广州": "小雨，气温 22-30°C，空气质量良",
    }
    return weather_data.get(city, f"未找到 {city} 的天气信息")

@tool
def calculator(expression: str) -> str:
    """计算数学表达式
    
    Args:
        expression: 数学表达式，如 2+3*4
    """
    try:
        allowed = set("0123456789+-*/().% ")
        if not all(c in allowed for c in expression):
            return "错误：表达式包含非法字符"
        return str(eval(expression))
    except Exception as e:
        return f"计算错误：{e}"

# 3. 工具列表
tools = [get_time, get_weather, calculator]

# 4. 获取 ReAct 提示词模板（从 LangChain Hub）
prompt = hub.pull("hwchase17/react")

# 5. 创建 Agent
agent = create_react_agent(llm, tools, prompt)

# 6. 创建执行器（对比昨天的 run() 方法）
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,  # 打印执行过程
    max_iterations=10,  # 对比昨天的 max_steps
    handle_parsing_errors=True  # 自动处理解析错误
)

# 7. 运行
result = agent_executor.invoke({"input": "北京天气怎么样？"})
print(f"\n最终答案: {result['output']}")

第二部分：对比分析

昨天的代码 vs LangChain
定义工具：
# 昨天（手写）
@registry.register(
    name="get_weather",
    description="获取指定城市的天气",
    parameters={"city": "城市名称"}
)
def get_weather(city: str) -> str:
    ...

# LangChain
@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气
    
    Args:
        city: 城市名称
    """
    ...
LangChain 从 docstring 自动提取描述和参数说明。

---
执行循环：
# 昨天（手写）- 约 50 行
def run(self, user_input: str, verbose: bool = True) -> str:
    for step in range(self.max_steps):
        response_text = self._call_llm(messages)
        parsed = self._parse_response(response_text)
        if parsed["final_answer"]:
            return parsed["final_answer"]
        if parsed["action"]:
            observation = self.registry.execute(action, **action_input)
            messages.append(...)
    ...

# LangChain - 1 行
result = agent_executor.invoke({"input": "问题"})

第三部分：自定义提示词

不想用 hub.pull()？自己写：

from langchain.prompts import PromptTemplate

custom_prompt = PromptTemplate.from_template("""你是一个有用的 AI 助手。

你可以使用以下工具：
{tools}

工具名称列表：{tool_names}

请严格按照以下格式回复：

Question: 用户的问题
Thought: 你的思考过程
Action: 工具名称
Action Input: 工具参数
Observation: 工具返回结果
... (可以重复多次 Thought/Action/Action Input/Observation)
Thought: 我现在知道最终答案了
Final Answer: 最终回答

开始！

Question: {input}
Thought: {agent_scratchpad}""")

agent = create_react_agent(llm, tools, custom_prompt)

第四部分：更现代的方式 - Tool Calling

OpenAI 的 Function Calling 比文本解析更可靠：
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

# 1. LLM（支持 tool calling）
llm = ChatOpenAI(model="gpt-4o-mini", api_key="你的key")

# 2. 工具（同上）
tools = [get_time, get_weather, calculator]

# 3. 提示词
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有用的助手，可以使用工具回答问题。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")  # 存放中间步骤
])

# 4. 创建 Agent（用 tool calling 而非文本解析）
agent = create_tool_calling_agent(llm, tools, prompt)

# 5. 执行器
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 6. 运行
result = agent_executor.invoke({"input": "计算 (15 + 27) * 3"})
print(result["output"])
区别：
- create_react_agent：LLM 输出文本，正则解析
- create_tool_calling_agent：LLM 输出结构化 JSON，更可靠

第五部分：获取执行历史

from langchain_core.messages import HumanMessage, AIMessage

# 方法 1：使用 return_intermediate_steps
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    return_intermediate_steps=True  # 关键参数
)

result = agent_executor.invoke({"input": "北京和上海天气对比"})

# 查看中间步骤
print("\n📜 执行历史:")
for i, (action, observation) in enumerate(result["intermediate_steps"]):
    print(f"\nStep {i+1}:")
    print(f"  Tool: {action.tool}")
    print(f"  Input: {action.tool_input}")
    print(f"  Output: {observation}")

print(f"\n最终答案: {result['output']}")

第六部分：错误处理

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10,
    handle_parsing_errors=True,  # 解析错误时自动重试
    # 或者自定义处理
    # handle_parsing_errors="请按正确格式重新回复"
)
自定义错误处理函数：
def custom_error_handler(error) -> str:
    return f"发生错误：{error}。请检查你的输出格式。"

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    handle_parsing_errors=custom_error_handler
)

第七部分：完整示例

from langchain_openai import ChatOpenAI
from langchain.agents import tool, create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
import datetime

# ========== 1. 配置 ==========
llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key="你的key",
    temperature=0
)

# ========== 2. 定义工具 ==========
@tool
def get_time() -> str:
    """获取当前时间"""
    return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气
    
    Args:
        city: 城市名称
    """
    data = {
        "北京": "晴，18-26°C",
        "上海": "多云，20-28°C",
        "广州": "小雨，22-30°C",
    }
    return data.get(city, f"未找到 {city} 的天气")

@tool
def calculator(expression: str) -> str:
    """计算数学表达式
    
    Args:
        expression: 数学表达式
    """
    try:
        return str(eval(expression))
    except:
        return "计算错误"

@tool
def search(query: str) -> str:
    """搜索信息
    
    Args:
        query: 搜索关键词
    """
    results = {
        "python": "Python 是流行的编程语言",
        "langchain": "LangChain 是 LLM 应用开发框架",
    }
    for key, value in results.items():
        if key in query.lower():
            return value
    return f"未找到 '{query}' 的相关信息"

tools = [get_time, get_weather, calculator, search]

# ========== 3. 创建 Agent ==========
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有用的助手。用中文回答。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

agent = create_tool_calling_agent(llm, tools, prompt)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10,
    return_intermediate_steps=True
)

# ========== 4. 测试 ==========
def run_test(query: str):
    print(f"\n{'='*60}")
    print(f"问题: {query}")
    print('='*60)
    
    result = agent_executor.invoke({"input": query})
    
    print(f"\n📜 执行步骤:")
    for i, (action, obs) in enumerate(result["intermediate_steps"]):
        print(f"  {i+1}. {action.tool}({action.tool_input}) → {obs}")
    
    print(f"\n✅ 答案: {result['output']}")

# 测试
run_test("现在几点了？")
run_test("北京和上海哪个更热？")
run_test("(100 + 50) * 2 等于多少？")
run_test("什么是 LangChain？")

框架 vs 手写 对比总结

今日检查清单

• 成功运行 LangChain Agent
• 理解 @tool 装饰器如何工作
• 理解 AgentExecutor 做了什么
• 知道 create_react_agent 和 create_tool_calling_agent 的区别
• 能获取并查看执行历史

扩展练习

练习 1：添加一个真实工具

import requests

@tool
def get_real_weather(city: str) -> str:
    """获取真实天气（需要 API key）
    
    Args:
        city: 城市名
    """
    # 使用免费的 wttr.in API
    response = requests.get(f"https://wttr.in/{city}?format=3")
    return response.text

练习 2：流式输出

for chunk in agent_executor.stream({"input": "北京天气"}):
    if "output" in chunk:
        print(chunk["output"], end="", flush=True)

明日预告

Day 16：LangChain 进阶 - Memory 与对话
让 Agent 记住对话历史：

• “刚才说的那个城市天气如何？”
• 多轮对话支持
• 不同类型的 Memory

---

有问题随时问！