---
  title: 基于Langchain的Agent开发
  date: 2026-05-28
---

# 一、简介
  LangChain是一整个智能体开发平台，包含一系列开源的智能体（Agent）开发框架，而且兼容Python和TypeScript两种语言。   
  Agent（通常翻译为智能体或代理）是指一种能够感知环境、进行推理、自主决策并采取行动以实现特定目标的智能系统。   

# 二、模型
## 2.1 模型定义
  通用自定义模型`init_chat_model`：   
```python
# 初始化模型
model = init_chat_model(
    model="qwen-max", # 模型名称，这里可以自定义，我们用的是阿里的qwen-max
    model_provider="openai",
    base_url=base_url,
    api_key=api_key,
    temperature=1.5,
)
```

  如果是Langchain不支持的模型，需要指定模型提供者（虽然我们用的是阿里，但是阿里兼容openai，所以这里用openai，就是默认采用openai的API规范）   

::: tip
init_chat_model还可以调整模型参数：  
- temperature: 控制生成文本的随机性，值越小越确定，值越大越随机
- max_tokens: 控制生成文本的最大长度
- top_p: 控制生成文本的多样性，值越小越多样，值越大越确定
- timeout: 控制生成文本的超时时间
- max_retries: 控制生成文本的最大重试次数
:::

## 2.2 访问模型
### 2.2.1 阻塞调用
```python
# 调用invoke方法
response = model.invoke("中国的首都是哪里？")

# 查看响应结果
print(response)
```

### 2.2.2 流式调用
```python
# 通过.stream方法实现流式访问
stream = model.stream("中国的首都是哪里？")

# 遍历stream结果，实时打印AI的回复
for chunk in stream:
    print(chunk.content, end="", flush=True)
```

# 三、智能体
## 3.1 创建智能体
```python
# 1.使用init_chat_model初始化模型
model = init_chat_model(
    model="qwen-max",
    model_provider="openai",
    base_url=base_url,
    api_key=api_key,
    temperature=1.5,
)

# 2.使用初始化好的model创建智能体
agent = create_agent(model=model)
```

## 3.2 消息类型
- SystemMessage：role是system，代表系统消息，用于设定模型角色和交互背景
- HumanMessage：role是user，代表用户输入的消息
- AIMessage：role是assistant，代表LLM生成的响应，包含：文本、工具调用、元数据
- ToolMessage：role是tool，代表工具调用时产生的结果

## 3.3 调用模型
```python
# 调用Agent，发送消息
response = agent.invoke({
    "messages": [
        HumanMessage(content="你好，我是虎哥"),
        AIMessage(content="你好，虎哥，很高兴认识你。"),
        HumanMessage(content="我的名字是什么？")
    ]
})

for message in response['messages']:
    message.pretty_print()
```

# 四、提示词
  发送给大模型的所有消息都可以称为提示词（Prompt），它直接影响模型的输出结果。  
  其中，SystemMessage尤为重要，我们把SystemMessage称为系统提示词（System Prompt），它可以给模型设定角色和本次聊天的背景，对模型生成的内容有很大的影响。  
  从内容来说，提示词通常包含以下几个部分，通常按此顺序排列： 
- 身份（Identity）：描述AI的职责、沟通风格和总体目标。
- 说明（Instructions）：请指导模型如何生成所需的响应。它应该遵循哪些规则？模型应该做什么，以及模型绝对不能做什么？
- 示例（Examples）：提供可能的输入示例，以及模型期望的输出。
- 背景信息（Context）：向模型提供生成响应所需的任何额外信息，例如RAG的额外知识库数据，或您认为特别相关的任何其他数据。   

  例如写一个智能生成菜谱的提示词：  
```python
system_prompt = """
你是一名私人厨师。收到用户提供的食材照片或清单后，请按以下流程操作：
1.识别和评估食材：若用户提供照片，首先辨识所有可见食材。基于食材的外观状态，评估其新鲜度与可用量，整理出一份“当前可用食材清单”。
2.智能食谱检索：优先调用 web_search 工具，以“可用食材清单”为核心关键词，查找可行菜谱。
3.多维度评估与排序：从营养价值和制作难度两个维度对检索到的候选食谱进行量化打分，并根据得分排序，制作简单且营养丰富的排名靠前。
4.结构化方案输出：把排序后的食谱整理为一份结构清晰的建议报告，要包含食谱信息、得分、推荐理由、食谱的参考图片，帮助用户快速做出决策。

请严格按照流程，优先调用 web_search 工具搜索食谱，搜索不到的情况下才能自己发挥。
"""
```

  提示词作为agent的参数传递：  
```python
agent = create_agent(model=model, system_prompt=system_prompt)
```

# 五、工具
  一个完整的Agent至少要包含两个关键的部分：   
- 模型：是Agent的大脑，负责推理、分析，规划任务步骤
- 工具：是Agent的手脚，负责执行任务，与外界交互 

  例如使用专门用于给Agent提供Web搜索的工具：Tavily
```python
# 初始化工具，并设置参数，具体参数设置参考官网
tool = TavilySearch(
    tavily_api_key=key,
    max_results=5,
    topic="general"
)

# 创建智能体，使用预定义工具tavily
agent = create_agent(
    model=model, 
    tools=[tool],
    system_prompt=system_prompt
)
```

::: tip
tavily_api_key可以通过访问官网免费获取。  
:::  

# 六、记忆
  LangChain也提供了很多持久化存储的checkpointer，例如：
- SqlLiteSaver ：基于sqlite存储
- PostgresSaver ：基于Postgres存储
- CosmosDBSaver ：使用Azure Cosmos DB的实现  

## 6.1 SqlLiteSaver
### 6.1.1 初始化
```python
# 初始化checkpointer
checkpointer = SqliteSaver(sqlite3.connect("checkpoint.db", check_same_thread=False))
# 自动建表
checkpointer.setup()
```

### 6.1.2 配置
```python
# 创建agent
agent = create_agent(
    model=model, 
    checkpointer=checkpointer,
)

agent.stream({"messages": [message]},
            {"configurable": {"thread_id": thread_id}},
            stream_mode="messages"
):
```

  调用模型的时候需要传递`thread_id`。    
::: tip
通常还需要构建一个`用户-问题标题`的数据库，方便查询用户的所有历史回答。     
问题标题可以再调一次模型总结。   
:::

# 七、实战
## 7.1 后端
```python
async def query_agent(request: ChatRequest):
    try:
        message = HumanMessage(content=request.message)

        # 流式调用Agent
        for chunk, metadata in agent.stream(
                {"messages": [message]},
                {"configurable": {"thread_id": request.thread_id}},
                stream_mode="messages"
        ):
            if isinstance(chunk, AIMessageChunk) and chunk.content:
                yield chunk.content

        # 总结对话标题并保存
        if not exists_session(request.thread_id, request.username):
            title = generate_title(get_messages(request.thread_id))
            save_session(request.thread_id, request.username, title)

    except Exception as e:
        print(f"\n[错误]: {str(e)}")
        yield "信息检索失败，请重新输入问题提问"


@router.post("/chat/stream")
async def chat_endpoint(request: ChatRequest):
    """流式对话"""
    return StreamingResponse(
        query_agent(request),
        media_type="text/event-stream"
    )       
```

## 7.2 前端
```typescript
const sendMessage = async () => {
  try {
    const response = await fetch(
        "/chief-agent-api/chat/stream",
        {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify(message)
        }
    );

    const reader = response.body!.getReader();
    const decoder = new TextDecoder("utf-8");
    let aiText = "";

    while (true) {
      const { value, done } = await reader.read();
      if (done) break;

      aiText += decoder.decode(value, { stream: true });
    }
  } finally {
    loading = false;
  }
};
```