Files

Cxx0822 b40d3f14a8 feat:增加TextToSql文档

2026-06-09 14:01:34 +08:00

8.0 KiB

Raw Blame History

title, date

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

一、简介

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

二、模型

2.1 模型定义

通用自定义模型init_chat_model：

# 初始化模型
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 阻塞调用

# 调用invoke方法
response = model.invoke("中国的首都是哪里？")

# 查看响应结果
print(response)

2.2.2 流式调用

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

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

三、智能体

3.1 创建智能体

# 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 调用模型

# 调用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的额外知识库数据，或您认为特别相关的任何其他数据。

例如写一个智能生成菜谱的提示词：

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

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

提示词作为agent的参数传递：

agent = create_agent(model=model, system_prompt=system_prompt)

五、工具

一个完整的Agent至少要包含两个关键的部分：

模型：是Agent的大脑，负责推理、分析，规划任务步骤
工具：是Agent的手脚，负责执行任务，与外界交互

例如使用专门用于给Agent提供Web搜索的工具：Tavily

# 初始化工具，并设置参数，具体参数设置参考官网
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 初始化

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

6.1.2 配置

# 创建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 后端

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 前端

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;
  }
};

8.0 KiB Raw Blame History Unescape Escape