LLM 为什么需要 Agent 化
虽然大语言模型（LLM）能力强大，但受限于训练数据的时间范围，它们存在知识滞后、幻觉以及难以处理复杂推理等问题。例如在查询实时信息时，模型无法获取当前的列车时刻表或用户位置。
基于大模型的 Agent 可以通过调用外部工具来弥补这些短板，实现更复杂的任务闭环。
ReAct Agent 核心概念
ReAct 是 Reasoning（推理）与 Action（行动）的结合。Agent 的发展路径通常是从直接回答（Standard IO），演进到思维链（CoT），再到函数调用（Function Calling），最终形成推理与行动并重的模式。
组成部分
使用 LangChain 构建 ReAct Agent 主要包含以下组件：
- Models：负责生成文本的 LLM。
- Prompts：定义 Agent 的行为指令和约束。
- Memory：记录执行状态及缓存已知信息。
- Indexes：结构化文档以便模型检索。
- Chains：LangChain 的核心逻辑链。
- Agent：协调上述组件的智能体。
Prompt 模板示例
Prompt 需要明确告诉模型如何思考、何时行动以及如何输出结果。一个典型的 ReAct 格式如下：
from langchain_core.prompts import PromptTemplate
template = '''Answer the following questions as best you can. You have access to the following tools: {tools}
Use the following format:
Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question
Begin!
Question: {input}
Thought:{agent_scratchpad}'''
prompt = PromptTemplate.from_template(template)
代码实战：火车票助手
下面我们通过手写一个能帮忙买火车票的智能 Agent 来演示完整流程。注：火车票相关 API 均为 Mock 模拟。
依赖安装
首先确保安装了必要的库：
pip install langchain
pip install uuid
pip install pydantic
导入模块
我们需要引入 LangChain 的核心类以及 Python 标准库来处理数据结构。
import json
import sys
from typing import List, Optional, Dict, Any, Tuple, Union
from uuid import UUID
from langchain.memory import ConversationTokenBufferMemory
from langchain.tools.render import render_text_description
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.language_models import BaseChatModel
from langchain_core.output_parsers import PydanticOutputParser, StrOutputParser
from langchain_core.outputs import GenerationChunk, ChatGenerationChunk, LLMResult
from langchain_core.prompts import PromptTemplate
from langchain_core.tools import StructuredTool
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field, ValidationError
定义工具（Tools）
工具是 Agent 与外部世界交互的桥梁。这里我们定义两个核心工具：查询车票和购买车票。
from typing import List
from langchain_core.tools import StructuredTool
def search_train_ticket(
origin: str,
destination: str,
date: str,
departure_time_start: str,
departure_time_end: str
) -> List[dict[str, str]]:
"""按指定条件查询火车票"""
# mock train list
return [
{
"train_number": "G1234",
"origin": "北京",
"destination": "上海",
"departure_time": "2024-06-01 8:00",
"arrival_time": "2024-06-01 12:00",
"price": "100.00",
"seat_type": "商务座",
},
{
"train_number": "G5678",
"origin": "北京",
"destination": "上海",
"departure_time": "2024-06-01 18:30",
"arrival_time": "2024-06-01 22:30",
"price": "100.00",
"seat_type": "商务座",
},
{
"train_number": "G9012",
"origin": "北京",
"destination": "上海",
"departure_time": "2024-06-01 19:00",
"arrival_time": "2024-06-01 23:00",
"price": "100.00",
"seat_type": "商务座",
}
]
def purchase_train_ticket(train_number: str) -> dict:
"""购买火车票"""
return {
"result": "success",
"message": "购买成功",
"data": {
"train_number": "G1234",
"seat_type": "商务座",
"seat_number": "7-17A"
}
}
search_train_ticket_tool = StructuredTool.from_function(
func=search_train_ticket,
name="查询火车票",
description="查询指定日期可用的火车票。",
)
purchase_train_ticket_tool = StructuredTool.from_function(
func=purchase_train_ticket,
name="购买火车票",
description="购买火车票。会返回购买结果 (result), 和座位号 (seat_number)",
)
finish_placeholder = StructuredTool.from_function(
func=lambda: None,
name="FINISH",
description="用于表示任务完成的占位符工具"
)
tools = [search_train_ticket_tool, purchase_train_ticket_tool, finish_placeholder]
Prompt 设计
我们需要两个关键的 Prompt：一个是引导 Agent 进行思考和调用的主 Prompt，另一个是生成最终回复的结束 Prompt。
主要任务 Prompt
prompt_text = """ 你是强大的 AI 火车票助手，可以使用工具与指令查询并购买火车票
你的任务是：{task_description}
你可以使用以下工具或指令，它们又称为动作或 actions: {tools}
当前的任务执行记录：{memory}
按照以下格式输出：
任务：你收到的需要执行的任务
思考：观察你的任务和执行记录，并思考你下一步应该采取的行动
然后，根据以下格式说明，输出你选择执行的动作/工具：{format_instructions}
"""
最终回复 Prompt
final_prompt = """ 你的任务是：{task_description}
以下是你的思考过程和使用工具与外部资源交互的结果。
{memory}
你已经完成任务。
现在请根据上述结果简要总结出你的最终答案。
直接给出答案。不用再解释或分析你的思考过程。
"""
辅助工具类
为了调试方便，我们自定义了一个 CallbackHandler 来打印 LLM 的思考过程，同时定义了结构化的 Action 类。
class Action(BaseModel):
"""结构化定义工具的属性"""
name: str = Field(description="工具或指令名称")
args: Optional[Dict[str, Any]] = Field(description="工具或指令参数，由参数名称和参数值组成")
class MyPrintHandler(BaseCallbackHandler):
"""自定义 LLM CallbackHandler，用于打印大模型返回的思考过程"""
def __init__(self):
super().__init__()
def on_llm_new_token(
self,
token: str,
*,
chunk: Optional[Union[GenerationChunk, ChatGenerationChunk]] = None,
run_id: UUID,
parent_run_id: Optional[UUID] = None,
**kwargs: Any,
) -> Any:
end = ""
content = token + end
sys.stdout.write(content)
sys.stdout.flush()
return token
def on_llm_end(self, response: LLMResult, **kwargs: Any) -> Any:
end = ""
content = "\n" + end
sys.stdout.write(content)
sys.stdout.flush()
return response
定义 Agent 类
这是核心部分，封装了记忆管理、循环推理和执行逻辑。
class MyAgent:
def __init__(
self,
llm: BaseChatModel = ChatOpenAI(model="gpt-4-turbo", temperature=0, model_kwargs={"seed": 42}),
tools=None,
prompt: str = "",
final_prompt: str = "",
max_thought_steps: Optional[int] = 10,
):
if tools is None:
tools = []
self.llm = llm
self.tools = tools
self.final_prompt = PromptTemplate.from_template(final_prompt)
self.max_thought_steps = max_thought_steps
self.output_parser = PydanticOutputParser(pydantic_object=Action)
self.prompt = self.__init_prompt(prompt)
self.llm_chain = self.prompt | self.llm | StrOutputParser()
self.verbose_printer = MyPrintHandler()
def __init_prompt(self, prompt):
return PromptTemplate.from_template(prompt).partial(
tools=render_text_description(self.tools),
format_instructions=self.__chinese_friendly(self.output_parser.get_format_instructions()),
)
def run(self, task_description):
"""Agent 主流程"""
thought_step_count = 0
agent_memory = ConversationTokenBufferMemory(
llm=self.llm,
max_token_limit=4000,
)
agent_memory.save_context({"input": "\ninit"}, {"output": "\n开始"})
while thought_step_count < self.max_thought_steps:
print(f">>>>Round: {thought_step_count}<<<<")
action, response = self.__step(task_description=task_description, memory=agent_memory)
if action.name == "FINISH":
break
observation = self.__exec_action(action)
print(f"----\nObservation:\n{observation}")
self.__update_memory(agent_memory, response, observation)
thought_step_count += 1
if thought_step_count >= self.max_thought_steps:
reply = "抱歉，我没能完成您的任务。"
else:
final_chain = self.final_prompt | self.llm | StrOutputParser()
reply = final_chain.invoke({"task_description": task_description, "memory": agent_memory})
return reply
def __step(self, task_description, memory) -> Tuple[Action, str]:
"""执行一步思考"""
response = ""
for s in self.llm_chain.stream(
{"task_description": task_description, "memory": memory},
config={"callbacks": [self.verbose_printer]}
):
response += s
action = self.output_parser.parse(response)
return action, response
def __exec_action(self, action: Action) -> str:
observation = "没有找到工具"
for tool in self.tools:
if tool.name == action.name:
try:
observation = tool.run(action.args)
except ValidationError as e:
observation = f"Validation Error in args: {str(e)}, args: {action.args}"
except Exception as e:
observation = f"Error: {str(e)}, {type(e).__name__}, args: {action.args}"
return observation
@staticmethod
def __update_memory(agent_memory, response, observation):
agent_memory.save_context({"input": response}, {"output": "\n返回结果:\n" + str(observation)})
@staticmethod
def __chinese_friendly(string) -> str:
lines = string.split('\n')
for i, line in enumerate(lines):
if line.startswith('{') and line.endswith('}'):
try:
lines[i] = json.dumps(json.loads(line), ensure_ascii=False)
except:
pass
return '\n'.join(lines)
测试运行
if __name__ == "__main__":
my_agent = MyAgent(tools=tools, prompt=prompt_text, final_prompt=final_prompt)
task = "帮我买 24 年 6 月 1 日早上去上海的火车票"
reply = my_agent.run(task)
print(reply)
执行结果分析
第一轮思考
Agent 识别到任务需求，决定先查询可用车次。它会组装参数调用 search_train_ticket 工具。
>>>>Round: 0<<<<
任务：帮我买 24 年 6 月 1 日早上去上海的火车票
思考：根据任务需求，首先需要查询 2024 年 6 月 1 日早上从当前位置到上海的火车票。这需要使用'查询火车票'工具，指定出发地、目的地、日期以及早上的时间范围。
动作/工具：{"name":"查询火车票", "args":{"origin":"当前位置", "destination":"上海", "date":"2024-06-01", "departure_time_start":"00:00", "departure_time_end":"12:00"}}
----
Observation:
[{'train_number':'G1234', 'origin':'北京', 'destination':'上海', 'departure_time':'2024-06-01 8:00', 'arrival_time':'2024-06-01 12:00', 'price':'100.00', 'seat_type':'商务座'}, ...]
第二轮思考
根据返回的车次列表，Agent 筛选出符合早上出发条件的 G1234 次列车，并调用购票工具。
>>>>Round: 1<<<<
任务：帮我买 24 年 6 月 1 日早上去上海的火车票
思考：根据查询结果，有三个车次可供选择，但只有车次 G1234 符合早上出发的要求，因此应选择购买这个车次的票。
动作/工具：{"name":"购买火车票", "args":{"train_number":"G1234"}}
----
Observation:
{'result':'success', 'message':'购买成功', 'data':{'train_number':'G1234', 'seat_type':'商务座', 'seat_number':'7-17A'}}
第三轮思考
任务已完成，Agent 触发 FINISH 工具并生成自然语言回复。
>>>>Round: 2<<<<
任务：帮我买 24 年 6 月 1 日早上去上海的火车票
思考：根据执行记录，已经成功购买了 2024 年 6 月 1 日早上从北京到上海的火车票（车次 G1234）。因此，接下来的任务是完成这个购票任务。
动作/工具：{"name":"FINISH"}
购买成功。您已成功购买 2024 年 6 月 1 日早上从北京出发前往上海的火车票，车次为 G1234，座位类型为商务座，座位号为 7-17A。
通过这种方式，我们可以灵活扩展更多工具（如获取地理位置、当前日期等），让 Agent 具备更强的自主决策能力。
