Chain与Agent实战

掌握了LangChain基础后，我们来深入学习Chain（链）和Agent（智能体）的高级用法。

Chain进阶

RAG链

检索增强生成（RAG）是最常用的链模式：

python

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_community.vectorstores import FAISS
from langchain_community.document_loaders import TextLoader

# 1. 加载文档并创建向量库
loader = TextLoader("knowledge.txt")
documents = loader.load()

from langchain_text_splitters import CharacterTextSplitter
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=100)
chunks = text_splitter.split_documents(documents)

embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(chunks, embeddings)
retriever = vectorstore.as_retriever()

# 2. 定义提示词
prompt = ChatPromptTemplate.from_messages([
    ("system", "根据以下上下文回答问题，如果上下文中没有相关信息，请说不知道：\n\n{context}"),
    ("user", "{question}")
])

# 3. 构建RAG链
model = ChatOpenAI(model="gpt-4o")

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

# 4. 执行
answer = rag_chain.invoke("什么是LangChain？")

对话链

带历史记忆的对话链：

python

from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables import RunnableWithMessageHistory

# 存储各会话的历史
store = {}

def get_session_history(session_id: str):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

# 对话提示词
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手"),
    ("placeholder", "{chat_history}"),
    ("user", "{input}")
])

chain = prompt | model | StrOutputParser()

# 包装成带历史的链
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history"
)

# 多轮对话
chain_with_history.invoke(
    {"input": "我叫张三"},
    config={"configurable": {"session_id": "user1"}}
)

chain_with_history.invoke(
    {"input": "我叫什么名字？"},
    config={"configurable": {"session_id": "user1"}}
)
# 输出：你叫张三

顺序链

按顺序执行多个步骤：

python

from langchain_core.runnables import RunnableParallel, RunnableLambda

# 第一步：翻译
translate_prompt = ChatPromptTemplate.from_template("翻译成英文：{text}")
translate_chain = translate_prompt | model | StrOutputParser()

# 第二步：总结
summarize_prompt = ChatPromptTemplate.from_template("用一句话总结：{text}")
summarize_chain = summarize_prompt | model | StrOutputParser()

# 组合：翻译 → 总结
full_chain = {
    "english": translate_chain,
    "original": RunnablePassthrough()
} | RunnableLambda(lambda x: summarize_chain.invoke({"text": x["english"]}))

result = full_chain.invoke({"text": "人工智能正在改变世界"})

Agent基础

什么是Agent？

Agent是能够自主决策、调用工具、多步完成任务的智能体：

普通Chain：输入 → 固定步骤 → 输出
Agent：输入 → 思考 → 选择工具 → 执行 → 观察 → 再思考 → ... → 输出

工具定义

python

from langchain_core.tools import tool

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气信息
    
    Args:
        city: 城市名称
        
    Returns:
        天气描述
    """
    # 模拟天气API
    weather_data = {
        "北京": "晴天，气温25°C",
        "上海": "多云，气温28°C",
        "广州": "小雨，气温30°C"
    }
    return weather_data.get(city, f"未找到{city}的天气信息")

@tool
def calculate(expression: str) -> str:
    """计算数学表达式
    
    Args:
        expression: 数学表达式，如 "2 + 3 * 4"
        
    Returns:
        计算结果
    """
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"计算错误：{e}"

tools = [get_weather, calculate]

创建Agent

python

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor

model = ChatOpenAI(model="gpt-4o", temperature=0)

# Agent提示词
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有用的助手，可以使用工具完成任务"),
    ("placeholder", "{chat_history}"),
    ("user", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

# 创建Agent
agent = create_tool_calling_agent(model, tools, prompt)

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

# 执行
result = agent_executor.invoke({"input": "北京今天天气怎么样？温度是多少？"})

Agent执行过程

> Entering new AgentExecutor chain...

思考：用户想知道北京天气，我需要调用get_weather工具
行动：get_weather(city="北京")
观察：晴天，气温25°C
思考：我现在知道了天气信息，可以回答用户了
最终答案：北京今天是晴天，气温25°C。

> Finished chain.

Agent高级用法

带记忆的Agent

python

from langchain_core.messages import AIMessage, HumanMessage

# 使用消息历史
from langchain.agents import AgentExecutor
from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    verbose=True
)

ReAct Agent

推理+行动模式的Agent：

python

from langchain.agents import create_react_agent

react_prompt = ChatPromptTemplate.from_messages([
    ("system", """你需要按以下格式回答问题：

Question: 用户的问题
Thought: 你应该思考做什么
Action: 要使用的工具名称
Action Input: 工具的输入
Observation: 工具的输出
... (这个过程可以重复多次)
Thought: 我现在知道最终答案了
Final Answer: 最终答案"""),
    ("user", "{input}\n{agent_scratchpad}")
])

agent = create_react_agent(model, tools, react_prompt)

自定义Agent行为

python

from langchain_core.runnables import RunnableConfig

# 配置Agent行为
config = RunnableConfig(
    max_concurrency=1,
    tags=["production"],
    metadata={"user_id": "user123"}
)

# 限制迭代次数
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=5,  # 最多5次工具调用
    handle_parsing_errors=True  # 处理解析错误
)

实战案例：智能问答Agent

python

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

# 定义工具
@tool
def search_knowledge(query: str) -> str:
    """搜索知识库"""
    # 这里可以接入真实的搜索系统
    return f"关于'{query}'的知识：..."

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        return str(eval(expression))
    except:
        return "计算失败"

@tool
def get_current_time() -> str:
    """获取当前时间"""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

# 创建Agent
tools = [search_knowledge, calculate, get_current_time]
model = ChatOpenAI(model="gpt-4o", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个智能助手，可以使用工具回答问题"),
    ("user", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 测试
questions = [
    "现在几点了？",
    "123 * 456 等于多少？",
    "什么是机器学习？"
]

for q in questions:
    result = agent_executor.invoke({"input": q})
    print(f"问：{q}\n答：{result['output']}\n")

小结

概念	说明
RAG Chain	检索增强生成链
对话链	带记忆的对话
Agent	自主决策的智能体
Tools	Agent可调用的工具
ReAct	推理+行动模式

下一步

继续学习记忆与工具详解，深入了解Agent的高级特性。

Chain与Agent实战 ​

Chain进阶 ​

RAG链 ​

对话链 ​

顺序链 ​

Agent基础 ​

什么是Agent？ ​

工具定义 ​

创建Agent ​

Agent执行过程 ​

Agent高级用法 ​

带记忆的Agent ​

ReAct Agent ​

自定义Agent行为 ​

实战案例：智能问答Agent ​

小结 ​

下一步 ​

Chain与Agent实战

Chain进阶

RAG链

对话链

顺序链

Agent基础

什么是Agent？

工具定义

创建Agent

Agent执行过程

Agent高级用法

带记忆的Agent

ReAct Agent

自定义Agent行为

实战案例：智能问答Agent

小结

下一步