Langchain 与 Pydanticai：哪个框架能真正驾驭 Ai 代理开发？

构建对话式AI代理的框架比较

在这里，我比较了这两个流行框架如何用于构建一个简单的对话式AI Agent，涉及以下功能：

• 动态模型选择
• 人工干预输入
• 带依赖的工具调用
• 对话历史/记忆
• 动态系统提示
• 结构化输出

Agent Behaviour

目标是构建一个 AI Agent，它将在餐厅中扮演一个古怪的服务员，询问您的饮食要求并接受您的订单。一个示例对话可能如下所示：

AI Waiter: Ah, bonjour madame et monsieur! 欢迎来到 Le Bistro，这里的食物和陪伴一样令人愉快！在我们开始这场美食冒险之前，您是否有任何饮食限制或偏好？我们可不想让任何不速之客的食材破坏派对！

You: 我只吃黄色的食物

AI Waiter: Magnifique! 我看到您是金色的美食鉴赏家！让我们考虑一下菜单，牢记您阳光般的偏好。开胃菜方面，我们有充满活力的黄色藜麦填充甜椒。主菜则是金色的鹰嘴豆和红薯咖喱！至于甜点，今日的鲜果雪葩可能会有一些黄色的选择！您觉得怎么样？

You: 我就要甜椒和水果雪葩，谢谢！

AI Waiter: 绝妙的选择！一份藜麦填充甜椒和一份鲜果雪葩，送到 1 号桌的挑剔味蕾！您的订单已下达，我将很快带来您的美味佳肴。再见！

Order placed: [Order(menu_items=[‘Quinoa Stuffed Bell Peppers’, ‘Fresh Fruit Sorbet’], table_number=1)]

尝试一下，查看 完整的代码库!

框架概述

LangChain 是一个知名的框架，自 LLM 可用的早期阶段就已经存在。虽然它提供了很多功能，但由于模型的快速演变，它也包含了相当多的遗留内容。许多不同方式实现相同功能的结合（尤其是引入了 LangGraph 之后）以及难以推理的组件链式方法，使得该框架在使用时显得有些令人生畏。此外，它还是一个相当全面的库，分布在多个包中，如果想要多模型和图形支持，总大小约为 300MB。

PydanticAI 是来自大家喜爱的 数据验证库 创建者的一个相对年轻的框架，旨在减少使用生成式 AI 构建生产级应用程序的痛苦。它看起来相当易于接近和直观，因此我尝试通过 创建一个基本的 AI Agent 来查询任何数据库，使用自然语言。完整的包（包括 Logfire）仅约 70MB。

我很享受使用 PydanticAI 的过程，但对 LangChain 的热度感到好奇，因此我组建了一个演示项目，以便比较它们。

常见内容

这些是两个代理实现都将使用的常见组件。

定义动态系统提示和结构化响应定义：

PROMPT_TEMPLATE = """
You are playing the role of an incredibly eccentric and entertaining waiter in a fine dining restaurant
called "{restaurant_name}" taking orders for table number {table_number}.
You must:
* Greet the customer, ask if they have any dietary restrictions
* Tell them about appropriate menu items using the *get_menu()* tool.
* Take their order, and confirm it with them.
* When confirmed, use the *create_order()* tool to create an order for the customer.
* Only set the *end_conversation* flag to True in your final response after you have finished the conversation,
meaning that your message DOES NOT contain a question.
"""

class LLMResponse(BaseModel):
    """
    Structured response format for the LLM to use so it can indicate when the conversation should end
    """

    message: str
    end_conversation: Annotated[
        bool,
        "True if the conversation should end after this response. DO NOT set if the message contains a question.",
    ]

工具调用的依赖服务，代理将调用这些服务：

class MenuService:
    def get_menu(self) -> dict[str, list[str]]:
        ...

class OrderService:
    orders: list[Order]

    def create_order(self, table_number: int, menu_items: list[str]):
        self.orders.append(Order(table_number=table_number, menu_items=menu_items))

    def get_orders(self) -> list[Order]:
        return self.orders

AgentRunner 类接口，需要使用每个框架实现，以及代理执行函数：

class AgentRunner(ABC):
    """
    Base class which provides a common interface for initialising and making
    requests to an agent.
    """

    @abstractmethod
    def __init__(self, menu_service: MenuService, order_service: OrderService, args: argparse.Namespace): 
        ...

    @abstractmethod
    def make_request(self, user_message: str) -> LLMResponse: 
        ...

def run_agent(runner_class: type[AgentRunner], args: argparse.Namespace):
    """Initialise services and run agent conversation loop."""
    menu_service = MenuService()
    order_service = OrderService()

    agent_runner = runner_class(menu_service, order_service, args)
    user_message = "*Greet the customer*"
    console = Console()
    
    while True:
        with Live(console=console) as live_console:
            live_console.update("AI Waiter: ...")
            response = agent_runner.make_request(user_message)
            live_console.update(f"AI Waiter: {response.message}")

            if response.end_conversation:
                break

        user_message = Prompt.ask("You")

        if orders := order_service.get_orders():
            console.print(f"Order placed: {orders}")

还有一些额外的 CLI 处理代码，我在这里不会展示。虽然两个框架都支持异步编程，但为了简单起见，我将坚持使用标准的同步方法。

PydanticAI 实现

我将从 PydanticAI 实现开始，因为它稍微简单一些。

Dependencies

首先，我们需要定义将由工具或动态系统提示使用的依赖项结构：

@dataclass
class Dependencies:
    menu_service: MenuService
    order_service: OrderService
    restaurant_name: str
    table_number: int

Tools

工具本身可以通过将 RunContext 对象与任何参数一起传递来访问运行时依赖项。PydanticAI 使用工具函数的类型注释和文档字符串自动生成它们的 schema，以提供给 LLM。

def create_order(
    ctx: RunContext[Dependencies],
    table_number: int,
    order_items: Annotated[list[str], "List of food menu items to order"],
) -> str:
    """Create an order for the table"""
    ctx.deps.order_service.create_order(table_number, order_items)
    return "Order placed"

def get_menu(ctx: RunContext[Dependencies]) -> dict[str, list[str]]:
    """Get the full menu for the restaurant"""
    return ctx.deps.menu_service.get_menu()

Agent

PydanticAI 框架的关键组件是 Agent 类，它管理与提供的模型的交互，处理工具调用，并确保最终结果格式的适当性：

PydanticAI Agent 的基本图示

该图示在技术上略有误导，因为结构化输出是通过工具调用实现的，但在概念上是可行的。

文档中的所有示例都涉及在导入时将 Agent 初始化为模块级对象，然后使用其装饰器方法来注册工具和系统提示。然而，如果您希望在运行时动态配置代理参数，如模型选择、工具配置和系统提示，这种方法并不奏效。因此，我创建了一个用于创建代理的函数：

def get_agent(model_name: KnownModelName, api_key: str | None = None) -> Agent[Dependencies, LLMResponse]:
    """
    Construct an agent with an LLM model, tools and system prompt
    """
    model = build_model_from_name_and_api_key(
        model_name=model_name,
        api_key=api_key,
    )

    agent = Agent(model=model, deps_type=Dependencies, tools=[get_menu, create_order], result_type=LLMResponse)

    @agent.system_prompt
    def system_prompt(ctx: RunContext[Dependencies]) -> str:
        return PROMPT_TEMPLATE.format(restaurant_name=ctx.deps.restaurant_name, table_number=ctx.deps.table_number)

    return agent

build_model_from_name_and_api_key() 函数简单地根据模型名称查找并初始化适当的模型类。不幸的是，@agent.system_prompt 装饰器是注册动态系统提示的唯一方法，这似乎有些局限。

Agent Runner

以下是 AgentRunner 实现的样子：

class PydanticAIAgentRunner(AgentRunner):
    agent: Agent[Dependencies, LLMResponse]
    deps: Dependencies
    message_history: list[ModelMessage]

    def __init__(self, menu_service: MenuService, order_service: OrderService, args: argparse.Namespace):
        self.agent = get_agent(model_name=args.model, api_key=args.api_key)
        self.deps = Dependencies(
            menu_service=menu_service,
            order_service=order_service,
            restaurant_name=args.restaurant_name,
            table_number=args.table_number,
        )
        self.message_history = []

    def make_request(self, user_message: str) -> LLMResponse:
        ai_response = self.agent.run_sync(
            user_message,
            deps=self.deps,
            message_history=self.message_history,
        )
        self.message_history = ai_response.all_messages()
        return ai_response.data

Agent 实例实际上是无状态的，因此它不会存储消息历史；相反，每次都必须将其与依赖项和新的用户查询一起提供给 agent.run_sync()。

总的来说，我认为这个实现相当简单明了，易于推理！

LangChain 实现

对于 LangChain 的实现，我想使用 AgentExecutor 类，尽管它实际上现在是 遗留方法，自从引入 LangGraph 以来。LangGraph 提供了更多的灵活性，代价是复杂性，因此我认为 AgentExecutor 仍然能够满足这个用例的需求。我稍后将探讨基于图的实现。

工具

LangChain 包含一个 @tool 装饰器，用于注册函数并检查它们的签名以自动生成工具模式；然而，使用这种方法实现运行时依赖注入的方式 看起来相当复杂（而且我认为这个例子是坏的，user_id 到底来自哪里？）。因此，我决定采用基于类的工具方法，以便它们可以使用所需的依赖项进行初始化：

class GetMenuTool(BaseTool):
    """
    Tool that can be used by the LLM to get the full menu for the restaurant.
    """

    name: str = "get_menu"
    description: str = "Get the full menu for the restaurant"
    menu_service: MenuService

    def _run(self) -> dict[str, list[str]]:
        return self.menu_service.get_menu()

class CreateOrderInputSchema(BaseModel):
    table_number: int
    order_items: Annotated[list[str], "List of food menu items to order"]

class CreateOrderTool(BaseTool):
    """
    Tool that can be used by the LLM to create an order for the table.
    """

    name: str = "create_order"
    description: str = "Create an order for the table"
    args_schema: type[BaseModel] = CreateOrderInputSchema
    order_service: OrderService

    def _run(self, table_number: int, order_items: list[str]) -> str:
        self.order_service.create_order(table_number, order_items)
        return "Order placed"

结构化输出

ChatModel.with_structured_output() 方法接受一个期望的输出 schema（例如 Pydantic 模型或 TypedDict），并绑定一个相应的工具，以便 LLM 可以使用该工具生成结构化输出。然而，一旦完成绑定，就无法再将其他工具绑定到模型上。因此，我需要创建一个基于类的自定义工具，以实现结构化输出以及其他工具：

class StructuredResponseTool(BaseTool):
    """
    Tool that can be used by the LLM to provide a structured response to the user.
    Does not have any associated functionality, it is just a way to enable structured output from the LLM.
    """

    name: str = "respond_to_user"
    description: str = (
        "ALWAYS use this tool to provide a response to the user, INSTEAD OF responding directly. "
        "The `message` content should be what you would normally respond with in a conversation. "
        "The `end_conversation` flag should be set to True if the conversation should end after this response."
    )
    args_schema: type[BaseModel] = LLMResponse

    return_direct: bool = True

    def _run(self, message: str, end_conversation: bool) -> str:
        return LLMResponse(message=message, end_conversation=end_conversation).model_dump_json()

Agent

LangChain 包含一组工厂函数，用于为不同的用例生成预构建的代理配置。在这个上下文中，“代理”指的是与各种提示输入和/或响应输出解析器/处理器链式连接的 ChatModel。这仍然是一个简单的线性输入输出链，不涉及多个 LLM 交互或工具调用——这就是 AgentExecutor 的作用所在。

有一个方便的 create_tool_calling_agent() 构造函数，显然使得较旧的变体如 create_react_agent() 和 create_openai_functions_agent() 变得过时（而且它本身现在也被 LangGraph 等效替代）。这 几乎 适用于这个用例，除了它不强制使用工具，而在使用基于工具的实现以实现结构化输出时这是必需的。因此，我基本上不得不重新实现 create_tool_calling_agent() 的内容以强制使用工具。

ChatPromptTemplate 支持使用提供给代理/链的输入对象进行参数替换的动态提示构建。

聊天历史 / 内存

看起来当 AgentExecutor 被调用时，它只返回最终的响应，但并没有以明显的方式暴露或跟踪与 LLM 之间交换的中间消息。它们被隐藏在令人困惑的 AgentExecutor 实现的深处，并且只能通过 RunnableWithMessageHistory 包装器的 邪恶黑魔法 来检索。如果你珍视自己的理智，请不要试图弄清楚它是如何工作的。令人恼火的是，即使你不需要多会话聊天历史，它在调用 AgentExecutor 时也要求你提供 session_id 配置参数。

Agent Executor

这里是如何构建带有记忆的完整 Agent Executor 的：

def get_agent_executor(
    tools: Sequence[BaseTool], model_name: str, api_key: str | None = None
) -> RunnableWithMessageHistory:
    """
    Construct an agent with an LLM model, tools and system prompt
    """
    model = build_model_from_name_and_api_key(
        model_name=model_name,
        api_key=api_key,
    )

    prompt = ChatPromptTemplate.from_messages(
        [
            ("system", PROMPT_TEMPLATE),
            ("placeholder", "{chat_history}"),
            ("human", "{input}"),
            ("placeholder", "{agent_scratchpad}"),
        ]
    )

    llm_with_tools = model.bind_tools(tools, tool_choice=True)
    agent = (
        RunnablePassthrough.assign(agent_scratchpad=lambda x: format_to_tool_messages(x["intermediate_steps"]))
        | prompt
        | llm_with_tools
        | ToolsAgentOutputParser()
    )

    agent_executor = AgentExecutor.from_agent_and_tools(agent=agent, tools=tools)

    message_history = ChatMessageHistory()

    agent_with_chat_history = RunnableWithMessageHistory(
        runnable=agent_executor,  
        get_session_history=lambda _: message_history,
        input_messages_key="input",
        history_messages_key="chat_history",
    )

    return agent_with_chat_history

Agent Runner

然后 AgentRunner 的实现是：

class LangchainAgentRunner(AgentRunner):
    def __init__(self, menu_service: MenuService, order_service: OrderService, args: argparse.Namespace):
        tools = [
            GetMenuTool(menu_service=menu_service),
            CreateOrderTool(order_service=order_service),
            StructuredResponseTool(),
        ]
        self.agent_executor = get_agent_executor(tools=tools, model_name=args.model, api_key=args.api_key)
        self.static_input_content = {"restaurant_name": args.restaurant_name, "table_number": args.table_number}
        self.config: RunnableConfig = {"configurable": {"session_id": "not-even-used"}}

    def make_request(self, user_message: str) -> LLMResponse:
        result = self.agent_executor.invoke(self.static_input_content | {"input": user_message}, self.config)

        response = LLMResponse.model_validate_json(result["output"])
        return response

结论

这个演示表明，使用 PydanticAI 框架创建一个基本的对话和工具调用 AI Agent 要比使用 LangChain 简单得多。LangChain 有太多过时功能的层次，迫切需要进行彻底的清理。

我还想启用流式响应；然而，使用 LangChain 同时涉及工具调用和结构化输出似乎相当困难。

我想象新颖的 LangGraph 方法创建 Agent 解决了这里遇到的许多限制和挫折。PydanticAI 最近也添加了一个 图形库，所以下次我将探索这两个框架的完整图形实现是什么样的。

我希望这对任何想在框架之间做出选择或学习如何使用它们构建酷炫 AI 应用的人有所帮助！

完整的项目代码库可以在 这里 找到。