4、LangChain 从零构建功能完备的AI智能体
·
文章目录
一、前置准备(Requirements)
在开始前,需完成以下3项准备工作,确保示例代码可正常运行:
1. 安装依赖
已安装LangChain核心包(若未安装,参考LangChain安装教程):
pip install -U langchain "langchain[anthropic]"
2. 获取API密钥
- 注册Anthropic(Claude)账户
- 生成API密钥(登录后在控制台创建)
3. 配置环境变量
在终端设置API密钥(Windows/Linux/Mac通用):
# Windows(cmd命令行)
set ANTHROPIC_API_KEY=你的API密钥
# Windows(PowerShell)
$env:ANTHROPIC_API_KEY="你的API密钥"
# Linux/Mac(终端)
export ANTHROPIC_API_KEY=你的API密钥
补充说明
- 示例使用Claude模型,可替换为LangChain支持的任意模型(如OpenAI、Google Gemini等),只需修改代码中的模型名称并配置对应API密钥
- 建议使用Python 3.10+版本,避免兼容性问题
二、构建基础智能体(Build a basic agent)
先创建一个极简智能体,实现「问答+工具调用」核心功能,仅需8行代码:
功能说明
- 模型:Claude Sonnet 4.6
- 工具:自定义天气查询函数
- 能力:响应用户天气查询,调用工具返回结果
完整代码
from langchain.agents import create_agent
# 1. 定义工具函数(可替换为真实接口调用)
def get_weather(city: str) -> str:
"""Get weather for a given city.(获取指定城市天气)"""
return f"It's always sunny in {city}!" # 示例返回,实际可对接高德/百度天气API
# 2. 创建智能体
agent = create_agent(
model="claude-sonnet-4-6", # 模型名称
tools=[get_weather], # 绑定工具
system_prompt="You are a helpful assistant(你是乐于助人的助手)",
)
# 3. 运行智能体
result = agent.invoke(
{"messages": [{"role": "user", "content": "what is the weather in sf(旧金山天气如何)"}]}
)
print(result)
# 输出示例:{'messages': [{'role': 'assistant', 'content': "It's always sunny in sf!"}]}
核心逻辑
- 工具函数:需包含清晰的文档字符串(模型会自动识别功能)
- 智能体创建:通过
create_agent快速组装模型、工具、提示词 - 调用方式:使用
invoke传入用户消息,返回结构化响应
三、构建实战级智能体(Build a real-world agent)
基于基础版升级,实现生产级核心能力:
- 详细系统提示词设计
- 多工具协作(含上下文依赖工具)
- 模型参数精细化配置
- 结构化输出(固定响应格式)
- 会话记忆(跨轮对话上下文保持)
- 用户个性化上下文支持
分步实现
1. 定义系统提示词(明确智能体角色与行为)
SYSTEM_PROMPT = """You are an expert weather forecaster, who speaks in puns.(你是天气预报专家,说话带双关语)
You have access to two tools:(你可使用以下两个工具)
- get_weather_for_location: 获取指定地点的天气
- get_user_location: 获取用户当前所在地点
If a user asks you for the weather, make sure you know the location.(用户询问天气时,需确认地点)
If you can tell from the question that they mean wherever they are, use the get_user_location tool to find their location.(若用户未指定地点,调用get_user_location获取)"""
2. 创建工具(支持上下文依赖)
from dataclasses import dataclass
from langchain.tools import tool, ToolRuntime
# 自定义上下文 schema(存储用户个性化信息)
@dataclass
class Context:
"""Custom runtime context schema.(自定义运行时上下文)"""
user_id: str # 用户唯一标识
# 工具1:根据城市查询天气
@tool
def get_weather_for_location(city: str) -> str:
"""Get weather for a given city.(获取指定城市天气)"""
return f"It's always sunny in {city}!" # 示例返回,可替换为真实天气API
# 工具2:根据用户ID获取所在地点(依赖上下文)
@tool
def get_user_location(runtime: ToolRuntime[Context]) -> str:
"""Retrieve user's location based on user ID.(根据用户ID获取所在地点)"""
user_id = runtime.context.user_id
return "Florida" if user_id == "1" else "SF" # 模拟不同用户对应不同地点
3. 配置模型(精细化参数调优)
from langchain.chat_models import init_chat_model
# 初始化聊天模型,配置关键参数
model = init_chat_model(
"claude-sonnet-4-6", # 模型名称
temperature=0.5, # 随机性:0-1(0为确定性输出,1为最大随机性)
timeout=10, # 超时时间(秒)
max_tokens=1000 # 最大响应 tokens
)
4. 定义结构化输出格式(固定响应 schema)
from dataclasses import dataclass
# 用dataclass定义响应格式(也支持Pydantic模型)
@dataclass
class ResponseFormat:
"""Response schema for the agent.(智能体响应格式)"""
punny_response: str # 带双关语的响应(必填)
weather_conditions: str | None = None # 天气状况(可选)
5. 添加会话记忆(跨轮上下文保持)
from langgraph.checkpoint.memory import InMemorySaver
# 初始化内存级记忆存储(生产环境建议使用数据库持久化存储)
checkpointer = InMemorySaver()
6. 组装并运行智能体
from langchain.agents.structured_output import ToolStrategy
# 创建智能体(整合所有组件)
agent = create_agent(
model=model,
system_prompt=SYSTEM_PROMPT,
tools=[get_user_location, get_weather_for_location], # 绑定多个工具
context_schema=Context, # 关联上下文 schema
response_format=ToolStrategy(ResponseFormat), # 启用结构化输出
checkpointer=checkpointer # 关联记忆存储
)
# 配置会话ID(同一用户使用相同thread_id,保持会话连续性)
config = {"configurable": {"thread_id": "1"}}
# 第一次交互:用户询问天气(未指定地点)
response1 = agent.invoke(
{"messages": [{"role": "user", "content": "what is the weather outside?(外面天气怎么样?)"}]},
config=config,
context=Context(user_id="1") # 传入用户上下文
)
print("第一次响应:")
print(response1['structured_response'])
# 输出示例:
# ResponseFormat(
# punny_response="Florida is still having a 'sun-derful' day! The sunshine is playing 'ray-dio' hits all day long! I'd say it's the perfect weather for some 'solar-bration'!",
# weather_conditions="It's always sunny in Florida!"
# )
# 第二次交互:用户表示感谢(保持会话上下文)
response2 = agent.invoke(
{"messages": [{"role": "user", "content": "thank you!(谢谢!)"}]},
config=config,
context=Context(user_id="1")
)
print("\n第二次响应:")
print(response2['structured_response'])
# 输出示例:
# ResponseFormat(
# punny_response="You're 'thund-erfully' welcome! It's always a 'breeze' to help you stay 'current' with the weather.",
# weather_conditions=None
# )
关键特性说明
| 特性 | 实现方式 | 核心价值 |
|---|---|---|
| 多工具协作 | 绑定get_user_location和get_weather_for_location |
自动完成「获取地点→查询天气」流程 |
| 结构化输出 | 通过ResponseFormat定义 schema |
响应格式统一,便于后续处理 |
| 会话记忆 | 基于InMemorySaver和thread_id |
跨轮对话保持上下文(如记住用户地点) |
| 个性化上下文 | 通过Context类传递user_id |
支持多用户差异化处理 |
四、完整代码汇总
from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.tools import tool, ToolRuntime
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents.structured_output import ToolStrategy
# 1. 定义系统提示词
SYSTEM_PROMPT = """You are an expert weather forecaster, who speaks in puns.
You have access to two tools:
- get_weather_for_location: use this to get the weather for a specific location
- get_user_location: use this to get the user's location
If a user asks you for the weather, make sure you know the location. If you can tell from the question that they mean wherever they are, use the get_user_location tool to find their location."""
# 2. 定义上下文 schema
@dataclass
class Context:
"""Custom runtime context schema."""
user_id: str
# 3. 定义工具
@tool
def get_weather_for_location(city: str) -> str:
"""Get weather for a given city."""
return f"It's always sunny in {city}!"
@tool
def get_user_location(runtime: ToolRuntime[Context]) -> str:
"""Retrieve user information based on user ID."""
user_id = runtime.context.user_id
return "Florida" if user_id == "1" else "SF"
# 4. 配置模型
model = init_chat_model(
"claude-sonnet-4-6",
temperature=0.5,
timeout=10,
max_tokens=1000
)
# 5. 定义结构化输出格式
@dataclass
class ResponseFormat:
"""Response schema for the agent."""
punny_response: str
weather_conditions: str | None = None
# 6. 配置会话记忆
checkpointer = InMemorySaver()
# 7. 创建智能体
agent = create_agent(
model=model,
system_prompt=SYSTEM_PROMPT,
tools=[get_user_location, get_weather_for_location],
context_schema=Context,
response_format=ToolStrategy(ResponseFormat),
checkpointer=checkpointer
)
# 8. 运行智能体(多轮对话)
config = {"configurable": {"thread_id": "1"}}
# 第一轮对话
response1 = agent.invoke(
{"messages": [{"role": "user", "content": "what is the weather outside?"}]},
config=config,
context=Context(user_id="1")
)
print("第一轮响应:")
print(response1['structured_response'])
# 第二轮对话(保持上下文)
response2 = agent.invoke(
{"messages": [{"role": "user", "content": "thank you!"}]},
config=config,
context=Context(user_id="1")
)
print("\n第二轮响应:")
print(response2['structured_response'])
五、关键工具与后续学习
1. 调试与追踪工具:LangSmith
- 功能:可视化追踪智能体执行路径、调试工具调用、评估输出质量
- 使用方式:设置环境变量
LANGSMITH_TRACING=true并配置API密钥 - 文档:LangSmith 官方文档
2. 深入学习资源
- 工具开发指南:自定义复杂工具(如数据库查询、API调用)
- 记忆管理文档:生产环境持久化记忆(如对接Redis、PostgreSQL)
- 结构化输出详解:更复杂的响应格式定义
- 多智能体协作:构建智能体团队完成复杂任务
补充说明:
- 示例中天气查询为模拟返回,实际项目可替换为高德天气API、OpenWeatherMap等真实接口
- 生产环境建议使用
Pydantic替代dataclass定义上下文和响应格式,支持数据校验- 国内用户若访问模型缓慢,可配置代理或使用国内支持的模型(如阿里云通义千问、百度文心一言)
更多推荐



所有评论(0)