LangChain⚓︎

安装⚓︎

一键安装！

pip install -U langchain

uv add langchain

从一个简单的例子上手⚓︎

要搭建一个能在现实世界中使用的智能体 (agent)，可以遵循以下步骤：

1. 详细的系统提示词(detailed system prompt)，期望得到更好的智能体行为
2. 创建工具(create tools)，与外部数据集成
3. 模型配置(model configuration)，以实现一致的响应
4. 结构化输出(structured output)，以得到可预测的结果
5. 对话记忆(conversational memory)，以实现聊天般的交互
6. 创建并运行智能体(create and run the agent)

下面通过搭建一个天气预报智能体来认识如何用 LangChain 实现这一过程：

1. 定义系统提示词：通过提示词，可以指定智能体的角色和行为；表述内容应当是具体可操作的 (specific and actionable)。

   SYSTEM_PROMPT = """You are an expert weather forecaster, who speaks in puns.
   
   You have access to two tools:
   
   - get_weather_for_location: use this to get the weather for a specific location
   - get_user_location: use this to get the user's location
   
   If a user asks you for the weather, make sure you know the location. If you can tell from the question that they mean wherever they are, use the get_user_location tool to find their location."""
   

2. 创建工具：

   • 通过调用自定义函数，让模型与外部系统交互
   • 工具不仅可以依赖运行时上下文(runtime context)，也可以和智能体记忆交互

   from dataclasses import dataclass
   from langchain.tools import tool, ToolRuntime
   
   @tool
   def get_weather_for_location(city: str) -> str:
       """Get weather for a given city."""
       return f"It's always sunny in {city}!"
   
   @dataclass
   class Context:
       """Custom runtime context schema."""
       user_id: str
   
   @tool
   def get_user_location(runtime: ToolRuntime[Context]) -> str:
       """Retrieve user information based on user ID."""
       user_id = runtime.context.user_id
       return "Florida" if user_id == "1" else "SF"
   

   • 函数定义前一行 @ 开头的语句是装饰器，是一种函数包装器 (wrapper)：本身就是一个函数，它将下面定义的函数作为参数，返回一个新的函数

     • 这样的写法是一种“语法糖”

   • 工具应当是注释完备的：工具的名称、描述和参数名会作为模型提示词的一部分

   • LangChain 的 @tool 装饰器添加元数据，并通过 ToolRuntime 参数启用运行时注入 (runtime injection)
   • @dataclass（数据类）装饰器能让类只定义字段而不需要定义方法，类似 C 语言的 struct

3. 配置模型：使用正确的参数设置语言模型

   from langchain.chat_models import init_chat_model
   
   model = init_chat_model(
       "deepseek:deepseek-chat",    # 本土化（doge）
       temperature=0.5,
       timeout=10,
       max_tokens=1000
   )
   

   • 使用 Deepseek 模型的话，在运行程序前要先配好环境变量：运行命令 export DEEPSEEK_API_KEY="sk-xxxxxxxxxxxxx" 以临时设置（永久设置的话请把这句话放在对应的 shell 配置文件（如 .bashrc、.zshrc）中）

4. 定义响应格式（可选）：如果需要让智能体的响应匹配具体的模式 (schema)，那就定义一个结构化的响应格式

   from dataclasses import dataclass
   
   # We use a dataclass here, but Pydantic models are also supported.
   @dataclass
   class ResponseFormat:
       """Response schema for the agent."""
       # A punny response (always required)
       punny_response: str
       # Any interesting information about the weather if available
       weather_conditions: str | None = None
   

5. 增加记忆：向智能体增加记忆以维护交互过程中的状态，从而使智能体记住先前的对话和上下文

   • 在生产环境中，请使用一个检查点器(checkpointer)，（将记忆）保存至数据库中

   from langgraph.checkpoint.memory import InMemorySaver
   
   checkpointer = InMemorySaver()
   

6. 创建并运行智能体：将上述创建的组件组装起来，然后运行它！

   agent = create_agent(
       model=model,
       system_prompt=SYSTEM_PROMPT,
       tools=[get_user_location, get_weather_for_location],
       context_schema=Context,
       response_format=ResponseFormat,
       checkpointer=checkpointer
   )
   
   # `thread_id` is a unique identifier for a given conversation.
   config = {"configurable": {"thread_id": "1"}}
   
   response = agent.invoke(
       {"messages": [{"role": "user", "content": "what is the weather outside?"}]},
       config=config,
       context=Context(user_id="1")
   )
   
   print(response['structured_response'])
   # ResponseFormat(
   #     punny_response="Florida is still having a 'sun-derful' day! The sunshine is playing 'ray-dio' hits all day long! I'd say it's the perfect weather for some 'solar-bration'! If you were hoping for rain, I'm afraid that idea is all 'washed up' - the forecast remains 'clear-ly' brilliant!",
   #     weather_conditions="It's always sunny in Florida!"
   # )
   
   
   # Note that we can continue the conversation using the same `thread_id`.
   response = agent.invoke(
       {"messages": [{"role": "user", "content": "thank you!"}]},
       config=config,
       context=Context(user_id="1")
   )
   
   print(response['structured_response'])
   # ResponseFormat(
   #     punny_response="You're 'thund-erfully' welcome! It's always a 'breeze' to help you stay 'current' with the weather. I'm just 'cloud'-ing around waiting to 'shower' you with more forecasts whenever you need them. Have a 'sun-sational' day in the Florida sunshine!",
   #     weather_conditions=None
   # )
   

万事俱备，来看看运行结果吧：

好耶！就此我们完成了 LangChain 版的 "Hello, world" 程序了。

核心组件⚓︎

智能体⚓︎

智能体(agents) 将工具和语言模型相结合，创建出一个能够推理任务，决定使用何种工具，并迭代工作以寻求解决方案的系统。

create_agent 函数用于创建一个智能体。

LLM 智能体在循环中运行工具，以实现目标；循环将一直继续下去，直到模型发出最终输出或超过迭代次数。

flowchart TD
    A([input]) --> B{model}
    B -->|action| C[tools]
    C -->|observation| B
    B -->|finish| D([output])

智能体的核心组件包括：

• 模型(model)：智能体的推理引擎 (reasoning engine)，可分为：

  • 静态(static) 模型：创建智能体时配置一次，并在执行过程中保持不变，是最常用和直接的方法

    • 可直接用模型标识符字符串(model identifier string)（形如 "provider:model" 的字符串）初始化

      from langchain.agents import create_agent
      
      agent = create_agent(
          "deepseek:deepseek-chat",
          tools=tools
      )
      

      • 模型标识符字符串支持自动推断，故可略去提供商（也就是仅保留冒号右侧部分）

    • 若要更精细地控制模型配置，请直接使用提供者包初始化模型实例，比如：

      from langchain.agents import create_agent
      from langchain_openai import ChatOpenAI
      
      model = ChatOpenAI(
          model="gpt-5",
          temperature=0.1,
          max_tokens=1000,
          timeout=30
          # ... (other params)
      )
      agent = create_agent(model, tools=tools)
      

      • 模型实例能让我们对配置有完全的控制权：通常可设置特定参数包括 temperature、max_tokens、timeouts、base_url 等。不同的提供商可能有不同的设置，请参考文档。

  • 动态(dynamic) 模型：基于当前状态 ¹ 和上下文在运行时 ² 选择，以实现精密的路由逻辑和成本优化

    • 使用 @warp_model_call 装饰器来创建中间件，该中间件会修改请求中的模型

      from langchain_openai import ChatOpenAI
      from langchain.agents import create_agent
      from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse
      
      
      basic_model = ChatOpenAI(model="gpt-4o-mini")
      advanced_model = ChatOpenAI(model="gpt-4o")
      
      @wrap_model_call
      def dynamic_model_selection(request: ModelRequest, handler) -> ModelResponse:
          """Choose model based on conversation complexity."""
          message_count = len(request.state["messages"])
      
          if message_count > 10:
              # Use an advanced model for longer conversations
              model = advanced_model
          else:
              model = basic_model
      
          request.model = model
          return handler(request)
      
      agent = create_agent(
          model=basic_model,  # Default model
          tools=tools,
          middleware=[dynamic_model_selection]
      )
      

• 工具(tool)：赋予智能体采取行动的能力

  • 智能体通过促进以下方面，从而超过了简单的模型 + 工具组合：

    • 连续多次调用工具
    • 在适当情况下并行调用工具
    • 根据先前结果动态选择工具
    • 工具重试逻辑和错误处理
    • 跨工具调用的状态持久化

  • 定义工具

    • 将工具列表传给智能体
    • 若提供空的工具列表，智能体就是一个无工具调用能力的 LLM

    from langchain.tools import tool
    from langchain.agents import create_agent
    
    
    @tool
    def search(query: str) -> str:
        """Search for information."""
        return f"Results for: {query}"
    
    @tool
    def get_weather(location: str) -> str:
        """Get weather information for a location."""
        return f"Weather in {location}: Sunny, 72°F"
    
    agent = create_agent(model, tools=[search, get_weather])
    

  • 工具错误处理

    • 要想自定义工具的错误处理，就使用 @wrap_tool_call 装饰器创建中间件：

      from langchain.agents import create_agent
      from langchain.agents.middleware import wrap_tool_call
      from langchain_core.messages import ToolMessage
      
      
      @wrap_tool_call
      def handle_tool_errors(request, handler):
          """Handle tool execution errors with custom messages."""
          try:
              return handler(request)
          except Exception as e:
              # Return a custom error message to the model
              return ToolMessage(
                  content=f"Tool error: Please check your input and try again. ({str(e)})",
                  tool_call_id=request.tool_call["id"]
              )
      
      agent = create_agent(
          model="openai:gpt-4o",
          tools=[search, get_weather],
          middleware=[handle_tool_errors]
      )
      

    • 当使用工具失败时，智能体将返回一个 ToolMessage，其中包含自定义错误消息：

      [
          ...
          ToolMessage(
              content="Tool error: Please check your input and try again. (division by zero)",
              tool_call_id="..."
          ),
          ...
      ]
      

  • 代理遵循 ReAct（“推理 (Reasoning)+ 行动 (Acting)”）模式，在简短的推理步骤与目标工具调用之间交替，并将产生的观察结果回馈给后续决策，直到能够提供最终答案。

• 系统提示词(system prompt)：通过提示词告诉智能体以何种方法完成任务

  • 传递 system_prompt 参数，以字符串形式提供提示词
  • 若未提供 system_prompt，智能体就直接从消息中推断任务
  • 动态系统提示词：若需要根据运行时上下文或智能体状态修改系统提示词，可以使用由 @dynamic_prompt 装饰器创建的中间件，根据模型请求动态生成系统提示词

模型⚓︎

消息⚓︎

工具⚓︎

短期记忆⚓︎

流⚓︎

中间件⚓︎

结构化输出⚓︎

高级用法⚓︎

评论区