1. 项目概述
LangChain作为当前最热门的LLM应用开发框架之一,其核心价值在于提供了构建智能Agent的标准范式。但很多开发者在从Demo过渡到实际应用时,往往会陷入概念迷宫。本文将以实战为导向,带你用LCEL(LangChain Expression Language)从零构建一个具备完整功能的自定义Agent。
1.1 核心需求解析
我们需要实现的Agent需要具备以下核心能力:
- 工具调用:能够根据用户需求选择合适的工具执行任务
- 多轮对话:支持上下文记忆和连贯的交互
- 错误处理:具备基本的容错和恢复机制
- 可观测性:提供详细的执行过程追踪
2. 环境准备与基础配置
2.1 开发环境搭建
首先确保你的Python环境版本≥3.8,然后安装必要的依赖包:
bash复制pip install langchain==0.1.0 langchain-openai==0.0.1 langchain-community==0.0.1 python-dotenv==1.0.0
注意:这里我们固定了主要包的版本号,因为LangChain的API在不同版本间可能有较大变化。实际开发中建议使用虚拟环境管理依赖。
2.2 API密钥配置
在项目根目录创建.env文件,添加你的OpenAI API密钥:
env复制OPENAI_API_KEY=sk-your-key-here
然后在代码中加载环境变量:
python复制from dotenv import load_dotenv
load_dotenv() # 自动加载.env文件
3. LCEL核心原理详解
3.1 声明式编程范式
LCEL(LangChain Expression Language)采用声明式编程范式,与传统的命令式编程相比有显著优势:
| 编程范式 | 特点 | 适用场景 |
|---|---|---|
| 命令式 | 详细描述执行步骤 | 简单线性流程 |
| 声明式 | 描述"做什么"而非"怎么做" | 复杂组合逻辑 |
LCEL的核心操作符是管道符|,它可以将多个Runnable组件串联起来,形成数据处理流水线。
3.2 Runnable接口规范
任何实现Runnable接口的组件都可以参与LCEL管道组合。Runnable接口定义如下:
python复制class Runnable(Protocol):
def invoke(self, input: Input) -> Output: ...
def stream(self, input: Input) -> Iterator[Output]: ...
def batch(self, inputs: List[Input]) -> List[Output]: ...
这种设计使得LCEL链天然支持:
- 同步调用(invoke)
- 流式输出(stream)
- 批量处理(batch)
4. 工具系统设计与实现
4.1 基础工具定义
工具是Agent的能力单元,每个工具需要明确定义三个要素:
- 名称(name):工具的调用标识符
- 描述(description):LLM决策的依据
- 执行函数(func):实际业务逻辑
使用装饰器定义简单工具:
python复制from langchain.tools import tool
@tool
def get_word_length(word: str) -> int:
"""返回输入单词的字符数量。当用户询问单词长度时使用此工具。"""
return len(word)
4.2 结构化工具开发
对于复杂参数的工具,推荐使用Pydantic模型定义参数结构:
python复制from langchain.tools import StructuredTool
from pydantic import BaseModel, Field
class CalculatorInput(BaseModel):
expression: str = Field(description="合法的Python数学表达式,如'2**10'")
def calculate(expr: str) -> str:
try:
return str(eval(expr, {"__builtins__": {}}, {}))
except Exception as e:
return f"计算错误:{e}"
calc_tool = StructuredTool.from_function(
func=calculate,
name="calculator",
description="执行数学计算",
args_schema=CalculatorInput
)
重要提示:工具描述的质量直接影响Agent的行为。好的描述应该包含:
- 工具的功能
- 使用场景
- 输入格式示例
- 可能的输出示例
5. 工具集(Toolkit)组织
5.1 内置工具集使用
LangChain提供了许多预置工具集,如数据库工具集:
python复制from langchain_community.agent_toolkits import SQLDatabaseToolkit
from langchain_community.utilities import SQLDatabase
db = SQLDatabase.from_uri("sqlite:///sales.db")
toolkit = SQLDatabaseToolkit(db=db)
tools = toolkit.get_tools() # 获取所有数据库相关工具
5.2 自定义工具集开发
创建自定义工具集类:
python复制from langchain.agents.agent_toolkits.base import BaseToolkit
from typing import List
class MathToolkit(BaseToolkit):
def get_tools(self) -> List[BaseTool]:
return [calc_tool, get_word_length]
6. Agent核心架构实现
6.1 Prompt工程
Agent的Prompt需要精心设计,通常包含以下部分:
python复制from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
("system", """你是一个智能助手,可以使用工具解决问题。
工具使用规范:
1. 每次调用工具前需说明推理过程
2. 工具输出会追加到对话历史"""),
("human", "{input}"),
MessagesPlaceholder("agent_scratchpad")
])
6.2 LCEL管道构建
使用LCEL组合各个组件:
python复制from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor
llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm_with_tools = llm.bind_tools(tools)
agent = (
{
"input": lambda x: x["input"],
"agent_scratchpad": lambda x: x["intermediate_steps"]
}
| prompt
| llm_with_tools
| OpenAIToolsAgentOutputParser()
)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=8
)
6.3 执行流程解析
AgentExecutor的工作流程:
- 接收用户输入
- 调用Agent进行决策
- 如果决定使用工具:
- 执行工具
- 将结果追加到中间步骤
- 重复步骤2
- 如果决定返回最终答案:
- 结束循环
- 返回结果
7. 记忆功能实现
7.1 对话历史管理
扩展Prompt支持多轮对话:
python复制prompt = ChatPromptTemplate.from_messages([
("system", "你是一个有记忆的助手"),
MessagesPlaceholder("chat_history"),
("human", "{input}"),
MessagesPlaceholder("agent_scratchpad")
])
7.2 历史记录维护
实现对话状态管理:
python复制from langchain_core.messages import AIMessage, HumanMessage
chat_history = []
def chat_round(user_input: str) -> str:
response = agent_executor.invoke({
"input": user_input,
"chat_history": chat_history
})
chat_history.extend([
HumanMessage(content=user_input),
AIMessage(content=response["output"])
])
return response["output"]
8. 生产环境优化
8.1 性能调优建议
- 工具并行化:对于独立工具调用,可以使用
ToolExecutor并行执行 - 缓存策略:为LLM和工具添加缓存层,减少重复计算
- 流式响应:利用LCEL原生支持的stream特性提升用户体验
8.2 监控与调试
集成LangSmith进行全链路追踪:
python复制import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "my_agent"
典型监控指标:
- 工具调用成功率
- 平均响应时间
- Token消耗统计
9. 完整实现代码
python复制import os
from dotenv import load_dotenv
from typing import List
from langchain.tools import tool, StructuredTool
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor
from langchain.agents.format_scratchpad.openai_tools import format_to_openai_tool_messages
from langchain.agents.output_parsers.openai_tools import OpenAIToolsAgentOutputParser
from langchain_core.messages import AIMessage, HumanMessage
from pydantic import BaseModel, Field
# 1. 环境配置
load_dotenv()
# 2. 工具定义
class CalculatorInput(BaseModel):
expression: str = Field(description="数学表达式,如'2**10'")
@tool
def calculate(expr: str) -> str:
"""执行数学计算,支持加减乘除和指数运算"""
try:
return str(eval(expr, {"__builtins__": {}}, {}))
except Exception as e:
return f"错误:{e}"
@tool
def get_time(timezone: str = "Asia/Shanghai") -> str:
"""获取指定时区的当前时间"""
from datetime import datetime
import pytz
return datetime.now(pytz.timezone(timezone)).strftime("%Y-%m-%d %H:%M:%S")
tools = [calculate, get_time]
# 3. LLM配置
llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm_with_tools = llm.bind_tools(tools)
# 4. Prompt模板
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个智能助手,可以使用工具解决问题"),
MessagesPlaceholder("chat_history"),
("human", "{input}"),
MessagesPlaceholder("agent_scratchpad")
])
# 5. Agent组装
agent = (
{
"input": lambda x: x["input"],
"chat_history": lambda x: x.get("chat_history", []),
"agent_scratchpad": lambda x: format_to_openai_tool_messages(
x["intermediate_steps"]
),
}
| prompt
| llm_with_tools
| OpenAIToolsAgentOutputParser()
)
# 6. Executor配置
executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=8,
handle_parsing_errors=True
)
# 7. 对话管理
history = []
def chat(user_input: str) -> str:
response = executor.invoke({
"input": user_input,
"chat_history": history
})
history.extend([
HumanMessage(content=user_input),
AIMessage(content=response["output"])
])
return response["output"]
# 8. 测试用例
if __name__ == "__main__":
print(chat("现在几点?"))
print(chat("3的5次方是多少?"))
print(chat("我之前问了什么问题?"))
10. 进阶扩展方向
- 工具组合:实现工具的串联调用,前一个工具的输出作为后一个工具的输入
- 动态工具加载:根据用户需求实时加载不同的工具集
- 验证机制:在工具调用前增加参数验证层
- 回退策略:当主要工具失败时自动尝试备用方案
在实际项目中,建议从简单场景开始,逐步增加复杂度。每次迭代后都要进行充分的测试,特别是边缘案例的测试。
