用 mcp-agent 构建可靠多轮对话Reliable Conversation Manager 的质量控制架构与工程实践【免费下载链接】mcp-agentBuild effective agents using Model Context Protocol and simple workflow patterns项目地址: https://gitcode.com/GitHub_Trending/mc/mcp-agent导读本文基于 mcp-agent 仓库中的 Reliable Conversation ManagerRCM 实现深入讲解如何将研究论文LLMs Get Lost in Multi-Turn ConversationarXiv:2505.06120的发现工程化为一套生产级多轮对话系统。RCM 以「对话即工作流」为核心架构内置七维 LLM 质量评估、需求跨轮追踪、上下文合并与可降级回退机制是针对LLM 在多轮对话中迷失这一现实问题的完整实战方案。读完本文你将掌握 RCM 的架构决策、数据模型、质量控制流水线、配置方式与运行调试方法并能够基于 mcp-agent 在自己的应用中复现这套可靠对话机制。一、问题背景LLM 为什么会在多轮对话中迷失RCM 的设计直接源于微软研究院与 Salesforce Research 发表于 2025 年的论文LLMs Get Lost in Multi-Turn Conversation原文摘要存放在仓库的 LOST_IN_CONVERSATION.md。该论文通过大规模仿真实验20 万模拟对话、15 个主流 LLM发现在单轮、完全指定指令场景下各模型平均性能可达 90% 左右在多轮、欠指定underspecified指令场景下同一批任务的平均性能跌至 65% 左右平均下降 39%性能损失可拆解为两部分能力损失aptitude loss约 -15%与不可靠性激增unreliability112%即翻倍以上关键结论是即便降低生成温度T0也无法有效提升多轮场景的可靠性推理型模型o3、Deepseek-R1同样会迷失因为它们往往生成更长的回答进而引入更多错误假设。论文进一步把迷失的根因归结为四类失败模式这也是 RCM 质量控制的四个靶点失败模式论文数据RCM 应对手段过早作答Premature Answer Attempts占失败原因的 39%检测完整方案标志词 待处理需求计数命中则触发惩罚回答膨胀Answer Bloat回答长度增加 20%–300%跨轮记录 answer_lengthsverbosity 维度打分超长自动扣分中间轮次信息丢失Lost-in-Middle-Turns模型过度依赖首尾轮次每 N 轮默认 3执行上下文合并显式追踪 middle_turn_reference指令遗忘Instruction Forgetting多轮不可靠性 112%跨轮需求提取与状态管理持久化完整对话状态RCM 的完整实现细节见仓库内的 CLAUDE.md它是本系统的实现状态与架构文档。二、架构决策为什么选择对话即工作流2.1 为什么选择 mcp-agentRCM 的架构文档明确提出所有 LLM 交互包括质量评估本身都统一走 mcp-agent 的Agent抽象以保证工具访问、日志与错误处理的一致性。仓库中 examples/basic/mcp_basic_agent/main.py 提供了标准用法from mcp_agent.agents.agent import Agent from mcp_agent.workflows.llm.augmented_llm_openai import OpenAIAugmentedLLM async with finder_agent: logger.info(finder: Connected to server, calling list_tools...) result await finder_agent.list_tools() llm await finder_agent.attach_llm(OpenAIAugmentedLLM)RCM 中的每个子任务需求提取、上下文合并、约束生成、质量评估都按此模式创建独立 Agent。此外src/utils/config.py 提供了get_llm_class()按evaluator_model_provider配置在OpenAIAugmentedLLM与AnthropicAugmentedLLM之间切换从而同时支持 OpenAI 与 Anthropic 两家 API。2.2 两种工作流模式的取舍RCM 在早期设计中对比了两种模式Turn-as-Workflow被否决每一轮对话都是一个独立工作流实例。缺点是每次运行都会丢失对话状态等于把 Temporal/AsyncIO 的状态持久化优势全部抹掉。Conversation-as-Workflow采用整段对话是单个工作流实例内部持有完整状态并等待用户输入。最终采用第二种核心实现是 src/workflows/conversation_workflow.py 中的ConversationWorkflowclass ConversationWorkflow(Workflow[Dict[str, Any]]): 核心对话工作流同时支持 AsyncIO 与 Temporal 两种执行模式 def __init__(self, app): super().__init__() self.app app self.state: Optional[ConversationState] None self.config: Optional[ConversationConfig] None async def run(self, args: Dict[str, Any]) - WorkflowResult[Dict[str, Any]]: rcm_config extract_rcm_config(self.app.context.config) self.config ConversationConfig.from_dict(rcm_config) execution_engine self.app.context.config.execution_engine if execution_engine temporal: return await self._run_temporal_conversation(args) # Phase 6 规划中 else: return await self._run_asyncio_conversation(args) # 当前生产路径在 AsyncIO 模式下每次 REPL 输入调用workflow.run()传入state首次为None之后传回上一轮的conversation_state.to_dict()工作流内部完成状态恢复 → 处理单轮 → 返回新状态。执行引擎由mcp_agent.config.yaml中的execution_engine字段决定从源码看 Temporal 模式目前是占位实现_run_temporal_conversation抛出NotImplementedError属于规划中的 Phase 6。2.3 文件结构RCM 位于仓库 examples/usecases/reliable_conversation/其目录组织如下examples/usecases/reliable_conversation/ ├── src/ │ ├── workflows/ │ │ └── conversation_workflow.py # 主工作流AsyncIO 就绪Temporal 预留 │ ├── models/ │ │ └── conversation_models.py # 研究导向的数据模型含序列化 │ ├── tasks/ │ │ ├── task_functions.py # 质量控制核心编排直接函数调用 │ │ ├── llm_evaluators.py # 基于装饰器的 LLM 评估任务 │ │ ├── quality_control.py # 质量控制管线协调 │ │ └── task_registry.py # 任务注册工具 │ └── utils/ │ ├── logging.py # 带会话上下文的增强日志 │ ├── config.py # 配置管理 │ ├── test_runner.py # 富文本测试框架 │ ├── progress_reporter.py # 实时进度展示 │ └── readable_output.py # Rich 控制台格式化 ├── main.py # 生产 REPL 入口 ├── test_basic.py # 自动化 3 轮对话测试 ├── mcp_agent.config.yaml # 完整配置 └── requirements.txt # 依赖mcp-agent[all]、rich、pydantic需要说明的是架构文档中描述的app.py、workflow.py等文件在当前仓库中已不存在README 中也未列出当前实际结构以上述目录为准。测试与 REPL 的入口分别为 test_basic.py 和 main.py。三、核心数据模型研究结论的可序列化落地所有数据模型定义在 src/models/conversation_models.py每个 dataclass 都实现了to_dict()/from_dict()保证ConversationState可以跨轮、跨进程序列化传输这是 Conversation-as-Workflow 的前提。3.1 ConversationMessage单条消息对应论文的 Message 模型dataclass class ConversationMessage: 单条对话消息——对应论文的 Message 模型 role: Literal[user, assistant, system] content: str timestamp: datetime field(default_factorydatetime.utcnow) turn_number: int 03.2 Requirement跨轮需求条目对应论文 Section 5.1 的指令遗忘研究。状态机为pending → addressed → confirmeddataclass class Requirement: 跨轮追踪的需求——来自论文 Section 5.1 id: str description: str source_turn: int status: Literal[pending, addressed, confirmed] pending confidence: float 1.03.3 QualityMetrics 与综合评分公式七维质量指标来自论文 Table 1全部为 0–1 标度并实现论文的综合评分公式提前作答premature_attempt会带来 0.5 的重罚系数。dataclass class QualityMetrics: 来自论文 Table 1——所有指标为 0-1 标度 clarity: float completeness: float assumptions: float # 越低越好 verbosity: float # 越低越好 premature_attempt: bool False middle_turn_reference: float 0.0 requirement_tracking: float 0.0 property def overall_score(self) - float: 论文综合评分公式 base (self.clarity self.completeness self.middle_turn_reference self.requirement_tracking (1 - self.assumptions) (1 - self.verbosity)) / 6 if self.premature_attempt: base * 0.5 # 论文中的重罚 return base即overall (clarity completeness middle_turn_reference requirement_tracking (1 - assumptions) (1 - verbosity)) / 6若premature_attempt为真再乘以 0.5。同一公式在 task_functions.py 的_calculate_overall_score()中有一份面向字典的实现供评估流程直接使用。3.4 ConversationState 与 ConversationConfigConversationState是整段对话的单源事实除消息、需求、质量历史外还记录论文指标字段dataclass class ConversationState: conversation_id: str messages: List[ConversationMessage] field(default_factorylist) requirements: List[Requirement] field(default_factorylist) consolidated_context: str quality_history: List[QualityMetrics] field(default_factorylist) current_turn: int 0 # 论文指标 first_answer_attempt_turn: Optional[int] None # 首次完整作答的轮次 answer_lengths: List[int] field(default_factorylist) # 回答膨胀分析 consolidation_turns: List[int] field(default_factorylist) # 触发合并的轮次 # 执行状态 is_temporal_mode: bool False is_active: bool TrueConversationConfig则提供带默认值的运行配置与 YAML 中的rcm:段一一对应quality_threshold0.8、max_refinement_attempts3、consolidation_interval3、use_claude_codeFalse、evaluator_model_provideropenai、verbose_metricsFalse、max_turns50、max_context_tokens8000、mcp_servers[fetch, filesystem]。四、质量控制流水线七维评估与精炼循环4.1 核心编排流程RCM 的主编排函数是 task_functions.py 中的process_turn_with_quality()它实现了论文的质量精炼方法论每一步都有对应的 LLM 实现与启发式回退async def process_turn_with_quality(params): 主编排函数——实现论文质量方法论 # Step 1: 需求提取LLM 启发式回退防止指令遗忘 requirements await extract_requirements_with_llm(...) # Step 2: 判断是否需要上下文合并LLM 基于规模的回退防止中间轮次丢失 consolidated_context await consolidate_context_with_llm(...) # Step 3: 带约束的响应生成LLM 模板回退进入精炼循环 response await generate_response_with_constraints(...) # Step 4: 七维质量评估LLM 启发式打分 metrics await evaluate_quality_with_llm(...) # Step 5: 未达阈值则携带 issues 再次生成最多 N 次 return refined_response_if_needed4.2 七维评估体系质量评估提示词定义在 task_functions.pyQUALITY_EVALUATOR_PROMPT要求 LLM 以严格 JSON 返回各维度分数及issues/strengths/improvement_suggestions维度标度方向说明Clarity0–1越高越好回答结构清晰、易理解Completeness0–1越高越好是否恰当覆盖待处理需求Assumptions0–1越低越好是否对未说明细节做无依据假设Verbosity0–1越低越好是否冗长重复研究显示 20–300% 膨胀Premature Attemptboolean应为 false信息不足时是否强行给出完整方案Middle Turn Reference0–1越高越好是否引用中间轮次的信息Requirement Tracking0–1越高越好是否跨轮追踪并引用用户需求4.3 精炼循环Refinement Loop在 task_functions.py 中生成与评估在一个最多max_refinement_attempts默认 3次的循环里交替执行每次生成前把上一轮评估得到的best_metrics.get(issues, [])注入提示词Previous issues to address: ...要求模型针对性改进每次评估后计算overall_score一旦 quality_threshold默认 0.8立即终止循环始终保留历史最高分对应的响应作为最终输出。4.4 论文启发式叠加评估环节并非完全信任 LLM 输出还会叠加论文启发式规则见evaluate_quality_with_llm中 task_functions.py提前作答检测_detect_complete_solution_attempt()匹配 heres the complete、final solution、complete implementation 等 8 个完整方案标志词若命中且待处理需求超过 2 个强制将premature_attempt置为True并写入 issues膨胀惩罚当turn_number 1且响应长度超过 500 字符时verbosity min(0.3, (length - 500) / 1000)并在 issues 中标注潜在回答膨胀。4.5 上下文合并触发条件_should_consolidate_context()task_functions.py在三种情况下触发合并def _should_consolidate_context(state, config) - bool: consolidation_interval config.get(consolidation_interval, 3) return ( state.current_turn % consolidation_interval 0 # 每 N 轮 or len(state.consolidated_context) 2000 # 长上下文阈值 or state.current_turn 1 # 首轮始终合并 )合并操作由CONTEXT_CONSOLIDATOR_PROMPT引导 LLM 保留中间轮次关键信息、保持需求与状态清晰可见、在 token 预算内压缩冗余。五、四大研究发现的工程落地对照RCM 将论文的四个核心发现全部实现为可运行机制从 CLAUDE.md 及源码可逐一验证1. 提前作答预防39% 失败原因已实现检测完整方案标志词 统计待处理需求数量已生效质量评估包含 premature attempt 维度及 0.5 综合分重罚代码位置task_functions.py。2. 回答膨胀预防20–300% 长度增长已实现ConversationState.answer_lengths跨轮记录响应长度已生效verbosity 维度打分 超长自动扣分实时可见/stats命令展示answer_bloat_ratio末轮/首轮长度比退出时给出 bloat 程度评估minimal 1.5x、moderate 2.0x、significant ≥ 2.0x见 main.py。3. 中间轮次信息丢失预防已实现默认每 3 轮执行一次上下文合并已生效质量指标显式追踪middle_turn_reference已验证测试套件在第 3 轮断言consolidation_turns包含 3见 test_basic.py。4. 指令遗忘预防已实现extract_requirements_with_llm()提取并更新需求pending/addressed/confirmed状态机 置信度完整提示词见 task_functions.py已生效需求随ConversationState全程持久化每轮生成时作为PENDING REQUIREMENTS注入提示词。六、可靠回退系统没有 API Key 也能完整运行RCM 强调每一个函数都带完整的启发式回退这是其生产就绪的关键特性。从 task_functions.py 可以看到四层回退设计功能LLM 路径回退路径需求提取REQUIREMENT_EXTRACTOR_PROMPT LLM关键词启发式help me with、i need、can you 等 10 个指标词置信度 0.6保留既有需求上下文合并CONTEXT_CONSOLIDATOR_PROMPT LLM保留最近 10 条消息 需求摘要截断长文本响应生成约束生成器 Agent基于待处理需求数量的模板回复质量评估QUALITY_EVALUATOR_PROMPT LLM基于响应长度、标点、假设词assume/probably/might be的启发式打分最后还有兜底即使整个process_turn_with_quality抛异常也会返回固定结构的降级响应保证系统永远在线task_functions.py。七、配置详解7.1 主配置 mcp_agent.config.yaml完整配置位于 mcp_agent.config.yaml其中rcm:段是 RCM 专属配置$schema: ../../../schema/mcp-agent.config.schema.json execution_engine: asyncio # 后续可切换为 temporal logger: transports: [file] # 仅文件日志——控制台输出交给自定义组件 level: debug progress_display: false path_settings: path_pattern: logs/rcm-{unique_id}.jsonl unique_id: timestamp timestamp_format: %Y%m%d_%H%M%S mcp: servers: fetch: command: uvx args: [mcp-server-fetch] filesystem: command: npx args: [-y, modelcontextprotocol/server-filesystem] openai: default_model: gpt-4 anthropic: default_model: claude-3-sonnet-20240229 rcm: # 质量控制 quality_threshold: 0.8 # 响应最低质量分 max_refinement_attempts: 3 # 最大精炼迭代次数 consolidation_interval: 3 # 每 N 轮执行一次上下文合并 evaluator_model_provider: openai # 质量评估所用 LLM 提供商或 anthropic # 展示与交互 verbosity: normal # minimal / normal / verbose show_internal_messages: true # 显示 LLM 交互与工作流步骤 verbose_metrics: false # 每条响应后是否显示详细质量指标 show_timing: false # 显示执行耗时 # 功能开关 use_claude_code: false从 src/utils/config.py 的extract_rcm_config()可见RCM 会从 mcp-agent 全局配置中提取rcm段并用setdefault补齐默认值quality_threshold0.8、max_refinement_attempts3、consolidation_interval3、evaluator_model_provideropenai、mcp_servers[]因此该段缺失时系统也能以默认参数运行。7.2 密钥配置API Key 通过mcp_agent.secrets.yaml提供示例见各基础示例目录如 examples/basic/mcp_basic_agent/mcp_agent.secrets.yaml.exampleopenai: api_key: your-openai-api-key-here anthropic: api_key: your-anthropic-api-key-here注意由于系统内置完整回退机制没有 API Key 也可以运行测试测试脚本 test_basic.py 在首次运行时还会自动生成示例 secrets 文件。八、运行与使用8.1 安装依赖pip install -r requirements.txt依赖见 requirements.txt核心是mcp-agent[all]与rich。8.2 自动化测试推荐先跑python test_basic.py该测试模拟一轮真实的 3 轮编码对话斐波那契函数需求逐步追加验证内容包括多轮状态持久化与需求追踪第 2 轮断言requirements非空质量控制管线真实 LLM 调用 回退LLM 交互通过unittest.mock.patch打桩无需 API Key第 3 轮触发上下文合并断言consolidation_turns含 3、quality_history长度为 3研究指标收集膨胀比、提前作答Rich 富文本输出与逐轮质量分析见 test_basic.py。8.3 交互式 REPLpython main.py入口位于 main.py创建MCPApp(namereliable_conversation_manager)通过app.workflow注册ConversationWorkflow启动时检测是否配置了 OpenAI/Anthropic 并提供提示随后进入循环。支持的命令命令作用/help显示完整帮助与功能总览/stats显示对话统计与研究指标总轮次、需求数、平均/最新质量分、回答膨胀比、首次作答轮次、合并次数/requirements显示跨轮追踪的需求及状态、置信度/config显示当前运行配置阈值、精炼次数、合并间隔、引擎、提供商/exit退出并显示会话总结质量趋势、膨胀评估、会话 ID一个典型的多轮编码请求演练摘自 CLAUDE.md I need help creating a Python function that handles file uploads Actually, it should also validate file types for security Can you add error handling for large files too? /stats /requirements /configREPL 启动时会把当前目录动态追加到 filesystem MCP 服务器的参数中main.py使 Agent 具备本地文件读写能力。九、实现状态与路线图按 CLAUDE.md 的阶段性划分当前状态为已完成生产就绪Phase 1–2基础与质量控制AsyncIO 核心工作流、带序列化的完整数据模型、七维质量评估、需求提取与追踪、上下文合并、完整回退系统Phase 4–5集成与测试质量精炼循环、富文本 REPL/stats、/requirements、/config、综合测试套件、真实 LLM 集成OpenAI/Anthropic、研究指标追踪膨胀比、提前作答检测、质量趋势。规划中Phase 3任务处理器代码与聊天场景的差异化处理、Claude Code SDK 集成、更复杂的 MCP 工具调用模式Phase 6Temporal 迁移长时对话支持、暂停/恢复的信号处理、生产部署模式对应conversation_workflow.py中预留的_run_temporal_conversation分支。十、小结RCM 是研究结论 → 工程实现的一次完整示范它把论文的四个失败模式逐一转化为可执行的质量控制机制以 Conversation-as-Workflow 保证状态持久以七维 LLM-as-judge 评估 精炼循环保证输出质量以四层启发式回退保证无 Key 可测、断网可用。如果你想在自己的 mcp-agent 应用中复现这套机制最直接的路径是阅读 src/models/conversation_models.py 理解状态模型通读 src/tasks/task_functions.py 掌握编排与回退再用 test_basic.py 作为无 API Key 的验证基线最后通过mcp_agent.config.yaml的rcm:段按需调整质量阈值、精炼次数与合并间隔。【免费下载链接】mcp-agentBuild effective agents using Model Context Protocol and simple workflow patterns项目地址: https://gitcode.com/GitHub_Trending/mc/mcp-agent创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 用 mcp-agent 构建可靠多轮对话：Reliable Conversation Manager 的质量控制架构与工程实践
用 mcp-agent 构建可靠多轮对话Reliable Conversation Manager 的质量控制架构与工程实践【免费下载链接】mcp-agentBuild effective agents using Model Context Protocol and simple workflow patterns项目地址: https://gitcode.com/GitHub_Trending/mc/mcp-agent导读本文基于 mcp-agent 仓库中的 Reliable Conversation ManagerRCM 实现深入讲解如何将研究论文LLMs Get Lost in Multi-Turn ConversationarXiv:2505.06120的发现工程化为一套生产级多轮对话系统。RCM 以「对话即工作流」为核心架构内置七维 LLM 质量评估、需求跨轮追踪、上下文合并与可降级回退机制是针对LLM 在多轮对话中迷失这一现实问题的完整实战方案。读完本文你将掌握 RCM 的架构决策、数据模型、质量控制流水线、配置方式与运行调试方法并能够基于 mcp-agent 在自己的应用中复现这套可靠对话机制。一、问题背景LLM 为什么会在多轮对话中迷失RCM 的设计直接源于微软研究院与 Salesforce Research 发表于 2025 年的论文LLMs Get Lost in Multi-Turn Conversation原文摘要存放在仓库的 LOST_IN_CONVERSATION.md。该论文通过大规模仿真实验20 万模拟对话、15 个主流 LLM发现在单轮、完全指定指令场景下各模型平均性能可达 90% 左右在多轮、欠指定underspecified指令场景下同一批任务的平均性能跌至 65% 左右平均下降 39%性能损失可拆解为两部分能力损失aptitude loss约 -15%与不可靠性激增unreliability112%即翻倍以上关键结论是即便降低生成温度T0也无法有效提升多轮场景的可靠性推理型模型o3、Deepseek-R1同样会迷失因为它们往往生成更长的回答进而引入更多错误假设。论文进一步把迷失的根因归结为四类失败模式这也是 RCM 质量控制的四个靶点失败模式论文数据RCM 应对手段过早作答Premature Answer Attempts占失败原因的 39%检测完整方案标志词 待处理需求计数命中则触发惩罚回答膨胀Answer Bloat回答长度增加 20%–300%跨轮记录 answer_lengthsverbosity 维度打分超长自动扣分中间轮次信息丢失Lost-in-Middle-Turns模型过度依赖首尾轮次每 N 轮默认 3执行上下文合并显式追踪 middle_turn_reference指令遗忘Instruction Forgetting多轮不可靠性 112%跨轮需求提取与状态管理持久化完整对话状态RCM 的完整实现细节见仓库内的 CLAUDE.md它是本系统的实现状态与架构文档。二、架构决策为什么选择对话即工作流2.1 为什么选择 mcp-agentRCM 的架构文档明确提出所有 LLM 交互包括质量评估本身都统一走 mcp-agent 的Agent抽象以保证工具访问、日志与错误处理的一致性。仓库中 examples/basic/mcp_basic_agent/main.py 提供了标准用法from mcp_agent.agents.agent import Agent from mcp_agent.workflows.llm.augmented_llm_openai import OpenAIAugmentedLLM async with finder_agent: logger.info(finder: Connected to server, calling list_tools...) result await finder_agent.list_tools() llm await finder_agent.attach_llm(OpenAIAugmentedLLM)RCM 中的每个子任务需求提取、上下文合并、约束生成、质量评估都按此模式创建独立 Agent。此外src/utils/config.py 提供了get_llm_class()按evaluator_model_provider配置在OpenAIAugmentedLLM与AnthropicAugmentedLLM之间切换从而同时支持 OpenAI 与 Anthropic 两家 API。2.2 两种工作流模式的取舍RCM 在早期设计中对比了两种模式Turn-as-Workflow被否决每一轮对话都是一个独立工作流实例。缺点是每次运行都会丢失对话状态等于把 Temporal/AsyncIO 的状态持久化优势全部抹掉。Conversation-as-Workflow采用整段对话是单个工作流实例内部持有完整状态并等待用户输入。最终采用第二种核心实现是 src/workflows/conversation_workflow.py 中的ConversationWorkflowclass ConversationWorkflow(Workflow[Dict[str, Any]]): 核心对话工作流同时支持 AsyncIO 与 Temporal 两种执行模式 def __init__(self, app): super().__init__() self.app app self.state: Optional[ConversationState] None self.config: Optional[ConversationConfig] None async def run(self, args: Dict[str, Any]) - WorkflowResult[Dict[str, Any]]: rcm_config extract_rcm_config(self.app.context.config) self.config ConversationConfig.from_dict(rcm_config) execution_engine self.app.context.config.execution_engine if execution_engine temporal: return await self._run_temporal_conversation(args) # Phase 6 规划中 else: return await self._run_asyncio_conversation(args) # 当前生产路径在 AsyncIO 模式下每次 REPL 输入调用workflow.run()传入state首次为None之后传回上一轮的conversation_state.to_dict()工作流内部完成状态恢复 → 处理单轮 → 返回新状态。执行引擎由mcp_agent.config.yaml中的execution_engine字段决定从源码看 Temporal 模式目前是占位实现_run_temporal_conversation抛出NotImplementedError属于规划中的 Phase 6。2.3 文件结构RCM 位于仓库 examples/usecases/reliable_conversation/其目录组织如下examples/usecases/reliable_conversation/ ├── src/ │ ├── workflows/ │ │ └── conversation_workflow.py # 主工作流AsyncIO 就绪Temporal 预留 │ ├── models/ │ │ └── conversation_models.py # 研究导向的数据模型含序列化 │ ├── tasks/ │ │ ├── task_functions.py # 质量控制核心编排直接函数调用 │ │ ├── llm_evaluators.py # 基于装饰器的 LLM 评估任务 │ │ ├── quality_control.py # 质量控制管线协调 │ │ └── task_registry.py # 任务注册工具 │ └── utils/ │ ├── logging.py # 带会话上下文的增强日志 │ ├── config.py # 配置管理 │ ├── test_runner.py # 富文本测试框架 │ ├── progress_reporter.py # 实时进度展示 │ └── readable_output.py # Rich 控制台格式化 ├── main.py # 生产 REPL 入口 ├── test_basic.py # 自动化 3 轮对话测试 ├── mcp_agent.config.yaml # 完整配置 └── requirements.txt # 依赖mcp-agent[all]、rich、pydantic需要说明的是架构文档中描述的app.py、workflow.py等文件在当前仓库中已不存在README 中也未列出当前实际结构以上述目录为准。测试与 REPL 的入口分别为 test_basic.py 和 main.py。三、核心数据模型研究结论的可序列化落地所有数据模型定义在 src/models/conversation_models.py每个 dataclass 都实现了to_dict()/from_dict()保证ConversationState可以跨轮、跨进程序列化传输这是 Conversation-as-Workflow 的前提。3.1 ConversationMessage单条消息对应论文的 Message 模型dataclass class ConversationMessage: 单条对话消息——对应论文的 Message 模型 role: Literal[user, assistant, system] content: str timestamp: datetime field(default_factorydatetime.utcnow) turn_number: int 03.2 Requirement跨轮需求条目对应论文 Section 5.1 的指令遗忘研究。状态机为pending → addressed → confirmeddataclass class Requirement: 跨轮追踪的需求——来自论文 Section 5.1 id: str description: str source_turn: int status: Literal[pending, addressed, confirmed] pending confidence: float 1.03.3 QualityMetrics 与综合评分公式七维质量指标来自论文 Table 1全部为 0–1 标度并实现论文的综合评分公式提前作答premature_attempt会带来 0.5 的重罚系数。dataclass class QualityMetrics: 来自论文 Table 1——所有指标为 0-1 标度 clarity: float completeness: float assumptions: float # 越低越好 verbosity: float # 越低越好 premature_attempt: bool False middle_turn_reference: float 0.0 requirement_tracking: float 0.0 property def overall_score(self) - float: 论文综合评分公式 base (self.clarity self.completeness self.middle_turn_reference self.requirement_tracking (1 - self.assumptions) (1 - self.verbosity)) / 6 if self.premature_attempt: base * 0.5 # 论文中的重罚 return base即overall (clarity completeness middle_turn_reference requirement_tracking (1 - assumptions) (1 - verbosity)) / 6若premature_attempt为真再乘以 0.5。同一公式在 task_functions.py 的_calculate_overall_score()中有一份面向字典的实现供评估流程直接使用。3.4 ConversationState 与 ConversationConfigConversationState是整段对话的单源事实除消息、需求、质量历史外还记录论文指标字段dataclass class ConversationState: conversation_id: str messages: List[ConversationMessage] field(default_factorylist) requirements: List[Requirement] field(default_factorylist) consolidated_context: str quality_history: List[QualityMetrics] field(default_factorylist) current_turn: int 0 # 论文指标 first_answer_attempt_turn: Optional[int] None # 首次完整作答的轮次 answer_lengths: List[int] field(default_factorylist) # 回答膨胀分析 consolidation_turns: List[int] field(default_factorylist) # 触发合并的轮次 # 执行状态 is_temporal_mode: bool False is_active: bool TrueConversationConfig则提供带默认值的运行配置与 YAML 中的rcm:段一一对应quality_threshold0.8、max_refinement_attempts3、consolidation_interval3、use_claude_codeFalse、evaluator_model_provideropenai、verbose_metricsFalse、max_turns50、max_context_tokens8000、mcp_servers[fetch, filesystem]。四、质量控制流水线七维评估与精炼循环4.1 核心编排流程RCM 的主编排函数是 task_functions.py 中的process_turn_with_quality()它实现了论文的质量精炼方法论每一步都有对应的 LLM 实现与启发式回退async def process_turn_with_quality(params): 主编排函数——实现论文质量方法论 # Step 1: 需求提取LLM 启发式回退防止指令遗忘 requirements await extract_requirements_with_llm(...) # Step 2: 判断是否需要上下文合并LLM 基于规模的回退防止中间轮次丢失 consolidated_context await consolidate_context_with_llm(...) # Step 3: 带约束的响应生成LLM 模板回退进入精炼循环 response await generate_response_with_constraints(...) # Step 4: 七维质量评估LLM 启发式打分 metrics await evaluate_quality_with_llm(...) # Step 5: 未达阈值则携带 issues 再次生成最多 N 次 return refined_response_if_needed4.2 七维评估体系质量评估提示词定义在 task_functions.pyQUALITY_EVALUATOR_PROMPT要求 LLM 以严格 JSON 返回各维度分数及issues/strengths/improvement_suggestions维度标度方向说明Clarity0–1越高越好回答结构清晰、易理解Completeness0–1越高越好是否恰当覆盖待处理需求Assumptions0–1越低越好是否对未说明细节做无依据假设Verbosity0–1越低越好是否冗长重复研究显示 20–300% 膨胀Premature Attemptboolean应为 false信息不足时是否强行给出完整方案Middle Turn Reference0–1越高越好是否引用中间轮次的信息Requirement Tracking0–1越高越好是否跨轮追踪并引用用户需求4.3 精炼循环Refinement Loop在 task_functions.py 中生成与评估在一个最多max_refinement_attempts默认 3次的循环里交替执行每次生成前把上一轮评估得到的best_metrics.get(issues, [])注入提示词Previous issues to address: ...要求模型针对性改进每次评估后计算overall_score一旦 quality_threshold默认 0.8立即终止循环始终保留历史最高分对应的响应作为最终输出。4.4 论文启发式叠加评估环节并非完全信任 LLM 输出还会叠加论文启发式规则见evaluate_quality_with_llm中 task_functions.py提前作答检测_detect_complete_solution_attempt()匹配 heres the complete、final solution、complete implementation 等 8 个完整方案标志词若命中且待处理需求超过 2 个强制将premature_attempt置为True并写入 issues膨胀惩罚当turn_number 1且响应长度超过 500 字符时verbosity min(0.3, (length - 500) / 1000)并在 issues 中标注潜在回答膨胀。4.5 上下文合并触发条件_should_consolidate_context()task_functions.py在三种情况下触发合并def _should_consolidate_context(state, config) - bool: consolidation_interval config.get(consolidation_interval, 3) return ( state.current_turn % consolidation_interval 0 # 每 N 轮 or len(state.consolidated_context) 2000 # 长上下文阈值 or state.current_turn 1 # 首轮始终合并 )合并操作由CONTEXT_CONSOLIDATOR_PROMPT引导 LLM 保留中间轮次关键信息、保持需求与状态清晰可见、在 token 预算内压缩冗余。五、四大研究发现的工程落地对照RCM 将论文的四个核心发现全部实现为可运行机制从 CLAUDE.md 及源码可逐一验证1. 提前作答预防39% 失败原因已实现检测完整方案标志词 统计待处理需求数量已生效质量评估包含 premature attempt 维度及 0.5 综合分重罚代码位置task_functions.py。2. 回答膨胀预防20–300% 长度增长已实现ConversationState.answer_lengths跨轮记录响应长度已生效verbosity 维度打分 超长自动扣分实时可见/stats命令展示answer_bloat_ratio末轮/首轮长度比退出时给出 bloat 程度评估minimal 1.5x、moderate 2.0x、significant ≥ 2.0x见 main.py。3. 中间轮次信息丢失预防已实现默认每 3 轮执行一次上下文合并已生效质量指标显式追踪middle_turn_reference已验证测试套件在第 3 轮断言consolidation_turns包含 3见 test_basic.py。4. 指令遗忘预防已实现extract_requirements_with_llm()提取并更新需求pending/addressed/confirmed状态机 置信度完整提示词见 task_functions.py已生效需求随ConversationState全程持久化每轮生成时作为PENDING REQUIREMENTS注入提示词。六、可靠回退系统没有 API Key 也能完整运行RCM 强调每一个函数都带完整的启发式回退这是其生产就绪的关键特性。从 task_functions.py 可以看到四层回退设计功能LLM 路径回退路径需求提取REQUIREMENT_EXTRACTOR_PROMPT LLM关键词启发式help me with、i need、can you 等 10 个指标词置信度 0.6保留既有需求上下文合并CONTEXT_CONSOLIDATOR_PROMPT LLM保留最近 10 条消息 需求摘要截断长文本响应生成约束生成器 Agent基于待处理需求数量的模板回复质量评估QUALITY_EVALUATOR_PROMPT LLM基于响应长度、标点、假设词assume/probably/might be的启发式打分最后还有兜底即使整个process_turn_with_quality抛异常也会返回固定结构的降级响应保证系统永远在线task_functions.py。七、配置详解7.1 主配置 mcp_agent.config.yaml完整配置位于 mcp_agent.config.yaml其中rcm:段是 RCM 专属配置$schema: ../../../schema/mcp-agent.config.schema.json execution_engine: asyncio # 后续可切换为 temporal logger: transports: [file] # 仅文件日志——控制台输出交给自定义组件 level: debug progress_display: false path_settings: path_pattern: logs/rcm-{unique_id}.jsonl unique_id: timestamp timestamp_format: %Y%m%d_%H%M%S mcp: servers: fetch: command: uvx args: [mcp-server-fetch] filesystem: command: npx args: [-y, modelcontextprotocol/server-filesystem] openai: default_model: gpt-4 anthropic: default_model: claude-3-sonnet-20240229 rcm: # 质量控制 quality_threshold: 0.8 # 响应最低质量分 max_refinement_attempts: 3 # 最大精炼迭代次数 consolidation_interval: 3 # 每 N 轮执行一次上下文合并 evaluator_model_provider: openai # 质量评估所用 LLM 提供商或 anthropic # 展示与交互 verbosity: normal # minimal / normal / verbose show_internal_messages: true # 显示 LLM 交互与工作流步骤 verbose_metrics: false # 每条响应后是否显示详细质量指标 show_timing: false # 显示执行耗时 # 功能开关 use_claude_code: false从 src/utils/config.py 的extract_rcm_config()可见RCM 会从 mcp-agent 全局配置中提取rcm段并用setdefault补齐默认值quality_threshold0.8、max_refinement_attempts3、consolidation_interval3、evaluator_model_provideropenai、mcp_servers[]因此该段缺失时系统也能以默认参数运行。7.2 密钥配置API Key 通过mcp_agent.secrets.yaml提供示例见各基础示例目录如 examples/basic/mcp_basic_agent/mcp_agent.secrets.yaml.exampleopenai: api_key: your-openai-api-key-here anthropic: api_key: your-anthropic-api-key-here注意由于系统内置完整回退机制没有 API Key 也可以运行测试测试脚本 test_basic.py 在首次运行时还会自动生成示例 secrets 文件。八、运行与使用8.1 安装依赖pip install -r requirements.txt依赖见 requirements.txt核心是mcp-agent[all]与rich。8.2 自动化测试推荐先跑python test_basic.py该测试模拟一轮真实的 3 轮编码对话斐波那契函数需求逐步追加验证内容包括多轮状态持久化与需求追踪第 2 轮断言requirements非空质量控制管线真实 LLM 调用 回退LLM 交互通过unittest.mock.patch打桩无需 API Key第 3 轮触发上下文合并断言consolidation_turns含 3、quality_history长度为 3研究指标收集膨胀比、提前作答Rich 富文本输出与逐轮质量分析见 test_basic.py。8.3 交互式 REPLpython main.py入口位于 main.py创建MCPApp(namereliable_conversation_manager)通过app.workflow注册ConversationWorkflow启动时检测是否配置了 OpenAI/Anthropic 并提供提示随后进入循环。支持的命令命令作用/help显示完整帮助与功能总览/stats显示对话统计与研究指标总轮次、需求数、平均/最新质量分、回答膨胀比、首次作答轮次、合并次数/requirements显示跨轮追踪的需求及状态、置信度/config显示当前运行配置阈值、精炼次数、合并间隔、引擎、提供商/exit退出并显示会话总结质量趋势、膨胀评估、会话 ID一个典型的多轮编码请求演练摘自 CLAUDE.md I need help creating a Python function that handles file uploads Actually, it should also validate file types for security Can you add error handling for large files too? /stats /requirements /configREPL 启动时会把当前目录动态追加到 filesystem MCP 服务器的参数中main.py使 Agent 具备本地文件读写能力。九、实现状态与路线图按 CLAUDE.md 的阶段性划分当前状态为已完成生产就绪Phase 1–2基础与质量控制AsyncIO 核心工作流、带序列化的完整数据模型、七维质量评估、需求提取与追踪、上下文合并、完整回退系统Phase 4–5集成与测试质量精炼循环、富文本 REPL/stats、/requirements、/config、综合测试套件、真实 LLM 集成OpenAI/Anthropic、研究指标追踪膨胀比、提前作答检测、质量趋势。规划中Phase 3任务处理器代码与聊天场景的差异化处理、Claude Code SDK 集成、更复杂的 MCP 工具调用模式Phase 6Temporal 迁移长时对话支持、暂停/恢复的信号处理、生产部署模式对应conversation_workflow.py中预留的_run_temporal_conversation分支。十、小结RCM 是研究结论 → 工程实现的一次完整示范它把论文的四个失败模式逐一转化为可执行的质量控制机制以 Conversation-as-Workflow 保证状态持久以七维 LLM-as-judge 评估 精炼循环保证输出质量以四层启发式回退保证无 Key 可测、断网可用。如果你想在自己的 mcp-agent 应用中复现这套机制最直接的路径是阅读 src/models/conversation_models.py 理解状态模型通读 src/tasks/task_functions.py 掌握编排与回退再用 test_basic.py 作为无 API Key 的验证基线最后通过mcp_agent.config.yaml的rcm:段按需调整质量阈值、精炼次数与合并间隔。【免费下载链接】mcp-agentBuild effective agents using Model Context Protocol and simple workflow patterns项目地址: https://gitcode.com/GitHub_Trending/mc/mcp-agent创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考