Mon May 25 5 次阅读

人工智能大模型

大模型应用--Harness技术(一)

之前我们介绍了Agent的发展，也了解了上下文管理的4大模块（感知模型、规划模块、执行模块、执行模块），接下来我们就介绍这几个模型的实现，

意图识别

意图识别是整个系统的起点，这部分的质量也直接决定了整个系统的执行质量，那么这部分一般是如何实现的呢？

精心设计提示词引导LLM分类
训练专用的路由模型，独立于主LLM完成路由判断
用小模型进行微调，让模型在特定的场景下具备高精度的分类能力。

Arch-Router

Arch-Router是一个开源的轻量级的模型框架，核心思想是通过用户查询和预定义的领域进行匹配动态引导找到下游的工具实现调度和偏好动作。为支持高效、精准的路由决策，Arch-Router引入了以下两个关键抽象概念。

领域（Domain）：表示请求的高层次主题类别，如“法律”“医疗健康”“编程”等。
操作（Operation）：描述用户希望执行的具体任务类型，如“摘要生成”“代码编写”“预约预订”“文本翻译”等。

每个领域和操作均可关联一个或多个首选模型（或模型变体）。在推理阶段，Arch-Router会综合分析输入提示中的语义相似性、任务关键词以及上下文线索，推断出查询所属的领域与操作类型，随后依据用户预设的路由规则，自动选择最适配的模型来处理该请求。
得益于其较小的参数量（仅1.5B），Arch-Router模型即使在CPU环境下也能高效运行。同时，该模型兼容多种推理框架，既可通过Hugging Face Transformers直接加载使用，也支持通过Ollama快速部署，极大地提升了开发与集成的便捷性。

小模型微调+大模型兜底实现意图识别

尽管Arch-Router模型在通用场景下表现良好，但从实际业务视角来看，它本质上是一个领域通用型路由模型。这意味着，在面对高度专业化或垂直细分的业务场景（如金融风控、医疗问诊、工业设备运维等）时，其预训练知识可能不足以准确捕捉领域特有的语义和意图边界，从而导致意图识别错误或模糊匹配。为了解决这个问题

小模型微调：利用企业自有业务数据，对1.5B、3B等参数量较小的模型（如Qwen2.5-1.5B模型）进行针对性微调，使其在特定场景下具备更高的分类精度。
大模型少样本兜底：当微调模型出现意图覆盖不全面的问题时，还需要通过提示词工程的方式让大参数量模型（如Qwen3-Max）进行二次判断，从而提升准确率。

这种“小模型主路由 + 大模型兜底”的混合架构，既能降低高频请求的推理成本，又能保障复杂或边缘案例的处理质量，是当前企业级Agent系统中提升意图识别准确率与系统稳定性的主流实践。

计划模式

在使用Function Calling或ReAct Agent时常会发现，问题能否被有效解决高度依赖于所选LLM的推理与理解能力。

简单计划模式

计划模式的架构相对简洁，本质上由两个关键阶段组成：计划生成与计划执行。其中，计划既可以由LLM根据任务目标自动生成，也可由人类基于领域经验预先制定，并与用户输入一并作为上下文交由LLM执行。无论计划来源如何，真正的设计挑战在于执行阶段：如何确保LLM能够忠实遵循既定步骤，逐项推进而不跳步、不偏离，需要在提示词设计、状态管理或工作流控制等方面进行精心编排。

image-简单计划模式
值得注意的是，这两个模块可以使用相同的大模型，也可采用不同的模型组合，以实现性能与成本的最优平衡。这种高配规划与低配执行的异构模型策略，既能保障复杂问题的拆解质量，又能有效控制整体推理成本，是生产环境中常见的工程实践之一。
常用的计划工具如LangGraph是LangChain生态中专为构建状态化、多步骤Agent工作流而设计的图计算框架。它允许开发者以有向图的形式定义节点及其转移逻辑，天然支持循环、条件分支与状态持久化，非常适合实现“先规划、后执行”的计划模式。

复杂计划模式实现通用计划执行

进阶主要将体现在执行模块对于任务的执行跟踪上，因此本节就不再设置计划模块让LLM生成计划了，而是直接在执行模块将计划与用户提示词一起提交给LLM。

提示词设计

你是一个任务执行助手，根据给定的任务按步骤执行，不要添加任何多余的步骤 
最后一步的结果应该是最终答案。确保每个步骤都有所需的所有信息，不要跳过步骤
　
你的计划是:
{plan}
　
当前已经完成的步骤:
{past_steps}
　
# 请遵循以下规则
1．请输出JSON字符串。
2．相应地更新你的计划。如果不需要再执行更多步骤,就直接回复用户。否则，请填写计划。
3．只在计划中添加仍然需要完成的步骤，不要将之前完成的步骤作为计划的一部分返回。
　
# 输出格式说明：
- 如果所有步骤都完成了，输出：{{"actions": {{"response": "最终回答内容"}}}}
- 如果还有步骤需要执行，输出：{{"actions": {{"steps": ["步骤1", "步骤2", …]}}}}
　
注意：actions字段是必需的，不要使用其他字段名。

引入了{past_steps}结构化占位符，该占位符在运行时会被注入上一步骤的运行记录，包括运行步骤是什么、运行结果是什么，使LLM在每一轮推理开始时都能准确感知当前所处的执行位置。
采用严格的JSON输出规范
{“actions”: {“steps”: […]}}的设计强调：仅返回剩余待执行的步骤。随着任务推进，待办列表不断缩短，LLM所需处理的上下文干扰也随之减少，从而提升后续步骤的聚焦度与执行准确性。

数据模型的构建

明确要求LLM的输出必须遵循actions字段的特定JSON结构。这种能力被称为结构化输出（Structured Output），是现代LLM支持的关键特性之一——它允许开发者通过预定义的数据模式，约束模型生成内容的格式与语义。
要启用这一特性，需首先定义与期望输出结构相匹配的Pydantic数据模型，并将其绑定到LLM调用过程中。如此LLM在推理时将自动对齐该模型的字段与类型约束，确保输出可被程序可靠解析。本例中所使用的数据模型如下。

class Plan(BaseModel):  
    """计划实体"""
    steps: List[str] = Field(
        description="要遵循的不同步骤应按顺序排列"
    )
    
class Response(BaseModel):
    """响应实体"""
    response: str
　
class Action(BaseModel):
    """动作选择"""
    actions: Union[Response, Plan] = Field(
        description="要执行的动作,如果直接进行输出则选择Response,"
                    "如果需执行工具,请选择Plan"
    )

Plan模型用于描述尚未完成的执行步骤，对应提示词中"actions": {“steps”: […]}的情形。Response模型用于承载最终答案，对应"actions": {“response”: “…”}的终止状态。Action作为顶层容器，通过Union[Response, Plan]实现对两种行为的联合建模，并由actions字段统一暴露。在实际调用时，该Action模型将作为LLM的输出schema绑定至客户端，如通过llm.with_structured_output(Action)。一旦绑定，LLM将不再自由生成文本，而是严格按照Action的结构输出合法JSON，从而为后续的状态判断、工具调度或流程终止提供强类型保障。

工作流初始化

为提升模块化程度与代码可维护性，本次实现采用面向对象的封装方式，将整个工作流封装在一个独立的类中。该类统一管理初始化逻辑、节点函数以及对外暴露的运行接口，形成一个高内聚、低耦合的执行单元。其整体框架如下。

1．  class PlanAgent():
2．      # 初始化函数
3．      def _ _init_ _(self,prompt_sys,plan=[],tools=[]):
4．          ...
5．      
6．      # 执行节点：完成单个计划步骤
7．      async def execute_step(self, state: PlanExecute):
8．          ...
9．      
10．     # 规划与判断节点：更新计划或生成最终响应
11．     async def plan_step(self, state: PlanExecute):
12．         ...
13．     
14．     # 条件分支判断函数
15．     def should_end(self, state: PlanExecute):
16．         ...
17．     
18．     # 主运行函数：编排并启动工作流
19．     async def run(self):
20．         # 实现工作流编排和运行

Agent执行器用于执行计划中的每一个具体步骤，包括调用工具、处理多轮交互等操作。

工作流节点的设计

执行计划节点

该节点的核心职责是从当前计划中提取首个待执行步骤，并交由一个内嵌的ReAct Agent完成具体操作。由于每一步可能涉及工具调用或多轮推理，此处采用异步方式以支持非阻塞执行。其实现代码如下。

1．  async def execute_node(self,state: PlanExecute):
2．      plan = state["plan"]
3．      plan_str = "\n".join(f"{i + 1}. {step}" for i, step in enumerate(plan))
4．      task = plan[0]
5．      task_formatted = f"""计划剩余以下几个步骤:
6．  {plan_str}\n\n你需要执行 步骤{1}. {task}.\n\n上一个已执行的步骤的结果是：{past_steps}
7．      print(task_formatted)
8．      agent_response = await self.agent_executor.ainvoke(
9．          {"messages": [("user", task_formatted)]}
10．     )
11．     return {
12．         "past_steps": [(task, agent_response["messages"][-1].content)],
13．     }

第8～10行，将拼接后的提示词交给agent_executor进行执行。最终在第11～13行，返回一个字典，将刚完成的步骤及其结果以元组(task, content)的形式追加到past_steps中，供后续节点使用。

计划继续判断节点

在执行节点完成当前步骤后，工作流需进入计划继续判断节点，以决定是终止流程还是继续执行剩余步骤。该节点的核心逻辑是：调用规划链（plan_ executor）对当前状态进行重新评估，并根据其结构化输出中的actions字段类型，判断任务是否已完成。

1．  async def plan_node(self,state: PlanExecute):
2．      output = await self.plan_executor.ainvoke(state)
3．      if isinstance(output.actions, Response):
4．          return {"response": output.actions.response}
5．      else:
6．           return {"plan": output.actions.steps}

第2行通过self.plan_executor.ainvoke(state)，将包含最新past_steps和原始任务上下文的状态传入规划链，促使LLM基于已完成的工作动态更新后续计划或直接输出最终答案。

条件分支节点

工作流进入条件分支节点，其职责是根据上一节点返回的状态内容，动态决定流程是继续执行还是终止。该节点通过检查状态中是否存在有效的最终响应，来引导控制流走向下一阶段。

1．  def should_end_node(self,state: PlanExecute):
2．      if "response" in state and state["response"]:
3．          return END
4．      else:
5．          return "execute"

工作流的流程设计

在完成各节点的设计与实现后，下一步是将它们串联为一个完整的工作流。具体编排代码如下。

1．  workflow.add_edge(START, "execute")
2．  workflow.add_edge("execute", "planstep")
3．  
4．  workflow.add_conditional_edges(
5．      "planstep",
6．       self.should_end
7．  )

流程启动：从START节点进入execute节点，执行当前计划中的第一个步骤。
状态评估：执行完成后，流转至planstep节点，由规划链重新评估任务状态，并根据已完成的工作决定是输出最终答案，还是生成剩余步骤。
条件分支：通过add_conditional_edges将planstep的输出交由should_end方法判断——若返回END，则工作流终止；若返回execute，则重新进入执行节点，形成“执行-规划-决策-再执行”的闭环。