基于 LangChain 的 ReAct AI Agent 搭建实战

定义工具（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 具备更强的自主决策能力。