LangChain高级：链式调用与Agent
--- title: "LangChain高级：链式调用与Agent" description: "深入掌握LangChain的高级Chain类型、Agent机制和自定义工具开发" tags: ["LangChain", "Chain", "Agent", "工具调用"] category: "llm" icon: "🧠"
LangChain高级：链式调用与Agent
Chain的类型
LangChain提供了多种Chain类型，适用于不同的应用场景。理解这些Chain的特点有助于选择合适的方案。
LCEL（LangChain Expression Language）
LCEL是LangChain推荐的链式调用方式，使用管道操作符组合组件：
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
# 基本LCEL链
chain = (
ChatPromptTemplate.from_template("用{style}风格解释{topic}")
| ChatOpenAI(model="gpt-3.5-turbo")
| StrOutputParser()
)
result = chain.invoke({"style": "简单", "topic": "量子计算"})
分支与并行
from langchain_core.runnables import RunnableParallel, RunnableLambda
# 并行执行多个链
parallel_chain = RunnableParallel(
summary=(
ChatPromptTemplate.from_template("总结：{text}")
| llm
| StrOutputParser()
),
keywords=(
ChatPromptTemplate.from_template("提取关键词：{text}")
| llm
| StrOutputParser()
)
)
result = parallel_chain.invoke({"text": "长文本内容..."})
# result = {"summary": "...", "keywords": "..."}
# 使用RunnablePassthrough传递数据
from langchain_core.runnables import RunnablePassthrough
chain = RunnableParallel(
enhanced_input=(
RunnablePassthrough.assign(
context=lambda x: retrieve_context(x["question"])
)
)
| prompt
| llm
)
条件分支
from langchain_core.runnables import RunnableBranch
# 根据条件选择不同的处理路径
def route(info):
if "数学" in info["topic"]:
return math_chain
elif "编程" in info["topic"]:
return code_chain
else:
return general_chain
branch = RunnableBranch(
route,
general_chain # 默认分支
)
Agent机制
Agent是LangChain中基于LLM的决策代理，可以根据用户输入动态选择工具：
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool
# 定义工具
@tool
def search_web(query: str) -> str:
"""搜索网页获取信息"""
# 实际实现调用搜索API
return f"搜索结果：关于{query}的信息..."
@tool
def calculator(expression: str) -> str:
"""计算数学表达式"""
try:
return str(eval(expression))
except:
return "计算错误"
# 创建提示词
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个有帮助的助手，可以使用工具来回答问题。"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}")
])
# 创建Agent
llm = ChatOpenAI(model="gpt-3.5-turbo")
tools = [search_web, calculator]
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True,
max_iterations=5
)
# 运行Agent
result = agent_executor.invoke({
"input": "北京今天天气怎么样？温度是多少度？"
})
print(result["output"])
自定义工具开发
使用@tool装饰器
from langchain_core.tools import tool
from pydantic import BaseModel, Field
# 简单工具
@tool
def get_weather(city: str) -> str:
"""获取指定城市的天气信息"""
# 实际实现调用天气API
return f"{city}今天晴，温度25度"
# 带参数验证的工具
class WeatherInput(BaseModel):
city: str = Field(description="城市名称")
unit: str = Field(default="celsius", description="温度单位")
@tool(args_schema=WeatherInput)
def get_weather_detailed(city: str, unit: str = "celsius") -> str:
"""获取指定城市的详细天气信息"""
temp = 25 if unit == "celsius" else 77
return f"{city}今天晴，温度{temp}{'°C' if unit == 'celsius' else '°F'}"
使用StructuredTool
from langchain_core.tools import StructuredTool
from typing import Optional
def multiply(a: int, b: int) -> int:
"""将两个数字相乘"""
return a * b
multiply_tool = StructuredTool.from_function(
func=multiply,
name="Multiply",
description="将两个数字相乘",
args_schema=None # 自动从函数签名推断
)
API工具
import requests
from langchain_core.tools import StructuredTool
from pydantic import BaseModel
class APIInput(BaseModel):
endpoint: str
method: str = "GET"
params: Optional[dict] = None
def call_api(endpoint: str, method: str = "GET", params: dict = None) -> str:
"""调用外部API"""
response = requests.request(method, endpoint, json=params)
return response.text
api_tool = StructuredTool(
name="API Caller",
description="调用外部API获取数据",
func=call_api,
args_schema=APIInput
)
Multi-Agent系统
LangChain支持构建多Agent协作系统：
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
# 创建多个专业化Agent
def create_agent(role, tools):
prompt = ChatPromptTemplate.from_messages([
("system", f"你是一个{role}专家。"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}")
])
agent = create_tool_calling_agent(llm, tools, prompt)
return AgentExecutor(agent=agent, tools=tools)
# 创建不同角色的Agent
research_agent = create_agent("研究", [search_web])
analysis_agent = create_agent("分析", [calculator])
writing_agent = create_agent("写作", [write_file])
# 协调者Agent
coordinator_prompt = ChatPromptTemplate.from_messages([
("system", """你是协调者，根据用户需求分配任务给其他Agent。
可用Agent: research, analysis, writing"""),
("human", "{input}")
])
# 简单的协调逻辑
def coordinate_task(task):
if "搜索" in task:
return research_agent.invoke({"input": task})
elif "计算" in task:
return analysis_agent.invoke({"input": task})
elif "写作" in task:
return writing_agent.invoke({"input": task})
else:
return research_agent.invoke({"input": task})
流式处理
# 流式输出Agent结果
async def stream_agent_response(query: str):
async for event in agent_executor.astream_events(
{"input": query},
version="v1"
):
kind = event["event"]
if kind == "on_chat_model_stream":
content = event["data"]["chunk"].content
if content:
print(content, end="", flush=True)
# 异步调用
import asyncio
asyncio.run(stream_agent_response("查询天气"))
掌握LangChain的Chain和Agent机制，可以构建出功能强大、灵活可扩展的LLM应用。