agentlanggraph工具调用国产模型工作流多模型路由

Agent 框架实战：用国产开源模型搭多步工作流

2026年11月7日by EXAIV Team

Agent 框架实战：用国产开源模型搭多步工作流

单轮问答只是 LLM 应用的起点。真正的价值在于让模型执行多步操作：查数据 → 分析结果 → 调用 API → 生成报告。这就是 Agent——一个围绕 LLM 构建的循环，让模型决定「下一步做什么」，框架负责执行并把结果反馈给模型。本文用 LangGraph 和国产开源模型搭一个可运行的 Agent 工作流。

30 秒版本

Agent = LLM + 工具循环 + 状态管理。模型收到任务 → 决定调什么工具 → 框架执行并回传结果 → 模型基于结果继续决策 → 循环直到给出最终回答。LangGraph 提供状态图管理；模型用 EXAIV 云端或本地部署都行。

Agent 的核心循环

一个 Agent 的最小工作单元是一个循环：

用户输入 → LLM 分析 → 决定调工具 → 执行工具 → 结果回传给 LLM → LLM 继续决策 → … → 最终回答

这个循环的关键是模型决定下一步做什么（而不是代码硬编码步骤），框架只负责「执行模型的决定并反馈结果」。

LangGraph 最小实现

LangGraph 用状态图（StateGraph）管理 Agent 的流转。以下是一个带工具调用的最小 Agent：

python

# pip install langgraph langchain-openai

from langgraph.graph import StateGraph, MessagesState, END
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI

# 用 EXAIV 端点接入国产模型
llm = ChatOpenAI(
    model="kimi-k3",  # 以 /models 目录实时为准
    api_key="你的APIKey",
    base_url="https://exaiv.com/api/ai/v1",
)

# 定义工具
def search_docs(query: str) -> str:
    """搜索知识库"""
    # 你的实现
    return f"搜索结果：{query}"

tools = {"search_docs": search_docs}

# Agent 节点
def agent_node(state):
    messages = state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}

# 工具执行节点
def tool_node(state):
    last_msg = state["messages"][-1]
    # 解析工具调用并执行
    results = []
    for call in last_msg.tool_calls or []:
        result = tools[call["name"]](**call["args"])
        results.append(f"工具 {call['name']} 返回：{result}")
    return {"messages": [{"role": "tool", "content": "\n".join(results)}]}

# 构建状态图
graph = StateGraph(MessagesState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")

# 条件路由：有工具调用就走工具节点，否则结束
def should_continue(state):
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return END

graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")  # 工具结果回到 Agent 继续
app = graph.compile()

# 运行
result = app.invoke({"messages": [HumanMessage(content="搜索关于 Qwen 的最新信息")]})
print(result["messages"][-1].content)

LangGraph 的价值

LangGraph 不是必需的——你可以手写 while 循环实现同样的逻辑。它的价值在于状态管理和条件路由的抽象：当 Agent 变复杂（多 Agent 协作、人工审批节点、并行分支）时，手写循环会失控，LangGraph 的状态图让复杂流程仍然可读、可调试。

多模型协作：不同步骤用不同模型

Agent 的不同步骤对模型的要求不同——这是国产开源模型组合使用的最佳场景：

python

# 编排决策用 GLM（工具调用稳定）
orchestrator = ChatOpenAI(
    model="glm-5.3-flash",
    api_key="你的APIKey",
    base_url="https://exaiv.com/api/ai/v1",
)

# 长文分析用 Kimi（长上下文）
reader = ChatOpenAI(
    model="kimi-k3",
    api_key="你的APIKey",
    base_url="https://exaiv.com/api/ai/v1",
)

# 多 Agent 协作
def analysis_agent(state):
    """读取文档并分析"""
    docs = state.get("documents", "")
    response = reader.invoke([HumanMessage(content=f"分析以下文档：\n{docs}")])
    return {"analysis": response.content}

def planning_agent(state):
    """基于分析结果做规划"""
    analysis = state.get("analysis", "")
    response = orchestrator.invoke([HumanMessage(content=f"基于分析做规划：\n{analysis}")])
    return {"plan": response.content}

最大轮次限制

Agent 循环必须有上限。没有限制的话，模型可能反复调用同一个工具而不会收敛。建议设 5-10 轮上限，超过后给模型注入「你已达到最大步数，请给出最终回答」的提示。

错误恢复的三个要点

1. 工具失败不是致命错误。 工具执行失败时把错误信息回传给模型，让它知道调用失败、可以尝试其他方式。不要让异常中断循环。

2. 超时降级。 如果某一步卡住（比如外部 API 无响应），设置超时并把「超时了」告知模型，而不是无限等待。

3. 结果大小控制。 工具返回的大结果（如整个数据库查询结果）要在回传给模型之前做摘要或截断，避免上下文爆炸。

从单 Agent 到多 Agent 的进阶路径

单 Agent 解决大多数场景。当你需要多个专门化的 Agent 协作时，进阶模式有：

模式	做法	适用
流水线	Agent A 的输出是 Agent B 的输入	文档分析 → 报告生成
分工协作	一个协调 Agent 分配任务给专业 Agent	复杂研究、代码审查
辩论	两个 Agent 给出相反意见再综合	决策验证、质量检查

不要过早进入多 Agent

多 Agent 增加了复杂度：状态同步、通信协议、失败传播。单 Agent 能解决的问题不要拆成多 Agent——这是 Agent 设计的最重要原则。另外，多 Agent 之间传递的信息也可能丢失语义或超出上下文限制，需要设计信息压缩和传递协议。

生产化前的检查清单

从 demo 进入生产前，至少检查这些项目：

工具权限：每个工具明确允许访问的资源、参数范围和调用身份，不要让模型拥有无限制的 shell 或数据库权限。
输入验证：工具参数由模型生成，必须像处理用户输入一样校验，尤其是 URL、文件路径、SQL 和外部 API 参数。
最大成本：设置每次任务的最大轮次、最大 token、最大工具调用数，防止异常循环消耗额度。
可观测性：记录每轮选择了什么工具、工具耗时、返回大小和最终结果，但注意不要把敏感内容写进无保护的日志。
人工接管：涉及发送邮件、修改生产数据、付费操作等不可逆动作时，加入人工确认节点。
回放测试：准备一组固定任务，每次更换模型或 Prompt 后重放，观察工具选择和最终结果是否退化。

Agent 的质量不是只看最终回答。工具选错、重试过多、上下文膨胀，都是需要单独监控的指标。

FAQ

LangGraph 和 LangChain 什么关系？

LangChain 提供基础组件（模型调用、工具定义、Prompt 模板），LangGraph 在其上提供状态图管理（多步流程、条件路由、循环）。用 RAG 用 LangChain，用 Agent 用 LangGraph，但两者可以共存。

国产模型能做好 Agent 吗？

能，但需要选对模型。GLM 系在工具调用稳定性上有积累，适合做编排决策；Kimi 系的长上下文适合需要大量阅读的分析步骤。关键是给模型配好工具定义和提示词。

Agent 循环的最大轮次设多少合适？

5-10 轮覆盖大多数场景。简单查询 2-3 轮就结束；复杂多步任务可能需要 5-8 轮。超过 10 轮通常意味着任务需要拆分，而不是让 Agent 无限跑。

本地模型能跑 Agent 吗？

可以。本地 GGUF 走 Ollama 的 OpenAI 兼容端点，LangGraph 的 ChatOpenAI 只改 base_url 即可。但本地模型的工具调用稳定性可能不如云端旗舰，需要更多测试。

Agent 和 RAG 什么关系？

RAG 可以是 Agent 工具集里的一个工具（搜索知识库），也可以独立使用。Agent 加上 RAG 工具，就变成「能查你的知识库再回答」的助手。两者是组合关系而非替代关系。

价格在哪里看？

Agent 的 token 消耗远高于单轮调用（循环多轮），实时价格见定价页，建议先用少量请求估算每任务的 token 消耗。

Agent 框架实战：用国产开源模型搭多步工作流

2026年11月7日by EXAIV Team

Agent 框架实战：用国产开源模型搭多步工作流

30 秒版本

Agent 的核心循环

一个 Agent 的最小工作单元是一个循环：

用户输入 → LLM 分析 → 决定调工具 → 执行工具 → 结果回传给 LLM → LLM 继续决策 → … → 最终回答

这个循环的关键是模型决定下一步做什么（而不是代码硬编码步骤），框架只负责「执行模型的决定并反馈结果」。

LangGraph 最小实现

LangGraph 用状态图（StateGraph）管理 Agent 的流转。以下是一个带工具调用的最小 Agent：

python

# pip install langgraph langchain-openai

from langgraph.graph import StateGraph, MessagesState, END
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI

# 用 EXAIV 端点接入国产模型
llm = ChatOpenAI(
    model="kimi-k3",  # 以 /models 目录实时为准
    api_key="你的APIKey",
    base_url="https://exaiv.com/api/ai/v1",
)

# 定义工具
def search_docs(query: str) -> str:
    """搜索知识库"""
    # 你的实现
    return f"搜索结果：{query}"

tools = {"search_docs": search_docs}

# Agent 节点
def agent_node(state):
    messages = state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}

# 工具执行节点
def tool_node(state):
    last_msg = state["messages"][-1]
    # 解析工具调用并执行
    results = []
    for call in last_msg.tool_calls or []:
        result = tools[call["name"]](**call["args"])
        results.append(f"工具 {call['name']} 返回：{result}")
    return {"messages": [{"role": "tool", "content": "\n".join(results)}]}

# 构建状态图
graph = StateGraph(MessagesState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")

# 条件路由：有工具调用就走工具节点，否则结束
def should_continue(state):
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return END

graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")  # 工具结果回到 Agent 继续
app = graph.compile()

# 运行
result = app.invoke({"messages": [HumanMessage(content="搜索关于 Qwen 的最新信息")]})
print(result["messages"][-1].content)

LangGraph 的价值

多模型协作：不同步骤用不同模型

Agent 的不同步骤对模型的要求不同——这是国产开源模型组合使用的最佳场景：

python

# 编排决策用 GLM（工具调用稳定）
orchestrator = ChatOpenAI(
    model="glm-5.3-flash",
    api_key="你的APIKey",
    base_url="https://exaiv.com/api/ai/v1",
)

# 长文分析用 Kimi（长上下文）
reader = ChatOpenAI(
    model="kimi-k3",
    api_key="你的APIKey",
    base_url="https://exaiv.com/api/ai/v1",
)

# 多 Agent 协作
def analysis_agent(state):
    """读取文档并分析"""
    docs = state.get("documents", "")
    response = reader.invoke([HumanMessage(content=f"分析以下文档：\n{docs}")])
    return {"analysis": response.content}

def planning_agent(state):
    """基于分析结果做规划"""
    analysis = state.get("analysis", "")
    response = orchestrator.invoke([HumanMessage(content=f"基于分析做规划：\n{analysis}")])
    return {"plan": response.content}

最大轮次限制

错误恢复的三个要点

1. 工具失败不是致命错误。 工具执行失败时把错误信息回传给模型，让它知道调用失败、可以尝试其他方式。不要让异常中断循环。

2. 超时降级。 如果某一步卡住（比如外部 API 无响应），设置超时并把「超时了」告知模型，而不是无限等待。

3. 结果大小控制。 工具返回的大结果（如整个数据库查询结果）要在回传给模型之前做摘要或截断，避免上下文爆炸。

从单 Agent 到多 Agent 的进阶路径

单 Agent 解决大多数场景。当你需要多个专门化的 Agent 协作时，进阶模式有：

模式	做法	适用
流水线	Agent A 的输出是 Agent B 的输入	文档分析 → 报告生成
分工协作	一个协调 Agent 分配任务给专业 Agent	复杂研究、代码审查
辩论	两个 Agent 给出相反意见再综合	决策验证、质量检查

不要过早进入多 Agent

生产化前的检查清单

从 demo 进入生产前，至少检查这些项目：

工具权限：每个工具明确允许访问的资源、参数范围和调用身份，不要让模型拥有无限制的 shell 或数据库权限。
输入验证：工具参数由模型生成，必须像处理用户输入一样校验，尤其是 URL、文件路径、SQL 和外部 API 参数。
最大成本：设置每次任务的最大轮次、最大 token、最大工具调用数，防止异常循环消耗额度。
可观测性：记录每轮选择了什么工具、工具耗时、返回大小和最终结果，但注意不要把敏感内容写进无保护的日志。
人工接管：涉及发送邮件、修改生产数据、付费操作等不可逆动作时，加入人工确认节点。
回放测试：准备一组固定任务，每次更换模型或 Prompt 后重放，观察工具选择和最终结果是否退化。

Agent 的质量不是只看最终回答。工具选错、重试过多、上下文膨胀，都是需要单独监控的指标。

FAQ

LangGraph 和 LangChain 什么关系？

国产模型能做好 Agent 吗？

Agent 循环的最大轮次设多少合适？

5-10 轮覆盖大多数场景。简单查询 2-3 轮就结束；复杂多步任务可能需要 5-8 轮。超过 10 轮通常意味着任务需要拆分，而不是让 Agent 无限跑。

本地模型能跑 Agent 吗？

可以。本地 GGUF 走 Ollama 的 OpenAI 兼容端点，LangGraph 的 ChatOpenAI 只改 base_url 即可。但本地模型的工具调用稳定性可能不如云端旗舰，需要更多测试。

Agent 和 RAG 什么关系？

价格在哪里看？

Agent 的 token 消耗远高于单轮调用（循环多轮），实时价格见定价页，建议先用少量请求估算每任务的 token 消耗。

Agent 框架实战：用国产开源模型搭多步工作流

Agent 框架实战：用国产开源模型搭多步工作流

Agent 的核心循环

LangGraph 最小实现

多模型协作：不同步骤用不同模型

错误恢复的三个要点

从单 Agent 到多 Agent 的进阶路径

生产化前的检查清单

FAQ

相关阅读

Agent 框架实战：用国产开源模型搭多步工作流

Agent 框架实战：用国产开源模型搭多步工作流

Agent 的核心循环

LangGraph 最小实现

多模型协作：不同步骤用不同模型

错误恢复的三个要点

从单 Agent 到多 Agent 的进阶路径

生产化前的检查清单

FAQ

相关阅读