解密prompt系列57. Agent Context Engineering - 多智能體代碼剖析

承接上篇對Context Engineering的探討，本文將聚焦多智能體框架中的上下文管理實踐。我們將深入剖析兩個代表性框架：字節跳動開源的基于預定義角色與Supervisor-Worker模式的 Deer-Flow ，以及在其基礎上引入動態智能體構建能力的清華CoorAgent。通過對它們設計思路和實現細節的拆解，提煉出多智能體協作中高效管理上下文的關鍵策略。

承接上篇對Context Engineering的探討，本文將聚焦多智能體框架中的上下文管理實踐。我們將深入剖析兩個代表性框架：字節跳動開源的基于預定義角色與Supervisor-Worker模式的 Deer-Flow ，以及在其基礎上引入動態智能體構建能力的清華 **CoorAgent ** 。通過對它們設計思路和實現細節的拆解，提煉出多智能體協作中高效管理上下文的關鍵策略。

Context Engineering Tips

從開源框架和我們日常的開發實踐中，優化多智能體上下文管理有以下一些思路。（注：只是一種Practice，并不是Best Practice，畢竟見山不是山嘛哈哈哈哈）

降低上下文長度與復雜度 (Reduce Length & Complexity)
- 上文丟棄 (Discard Context)： 如Deer-Flow的Coordinator。若用戶提問與智能體任務無關，在回復用戶后，清空歷史對話消息 (Message History = 0)。
- 上文隔離 (Isolate Context)： Supervisor-Worker模式的核心。每個Worker智能體擁有獨立的上文環境，僅接收自身任務目標作為啟動指令，與Supervisor及其他Worker隔離。
- 上文分解 (Decompose Context)： Supervisor（Planner）將復雜任務拆解為獨立步驟。每個步驟復雜度降低，對應負責Worker的上文也隨之簡化。
降低上下文噪音 (Reduce Context Noise)
- 關注結果而非過程 (Focus on Results)： 關鍵策略！尤其適用于RePlan、Reflection等判斷模塊。無需傳遞冗長的執行過程消息，只需基于當前結果與目標狀態即可進行后續規劃或反思。直接傳遞原始Message往往是便捷但非最優解。
- 過濾/選擇上文 (Filter/Select Context)： 類比RAG中的Re-ranking。例如Coding Agent中，子Agent僅回傳關鍵stdout而非完整代碼。
- 壓縮上文 (Compress Context)： 如Deer-Flow中，Researcher對搜索步驟結果進行摘要；Reporter僅基于壓縮后的摘要而非原始引用進行最終總結。在Coding Agent設計中也很常見。

框架1：Deer-Flow

Deer Flow

Deer Flow是字節跳動推出的多智能體框架，后面的很多agent框架里都能看到它的影子，Deer Flow同樣使用Langgraph構建，包含以下幾個核心節點

協調者節點 (coordinator): 負責與用戶交互，判斷是否需要轉交給規劃者
核心規劃模塊（Planner）：負責生成詳細的任務執行計劃
- 前置背景調查節點 (background_investigator): 執行初步網絡搜索獲取背景信息
- 后置人工反饋節點 (human_feedback): 允許用戶審查和修改計劃
研究團隊節點 (research_team): 協調研究者和編碼者
- 研究者節點 (researcher): 執行信息收集和研究任務
- 編碼者節點 (coder): 執行數據處理和代碼分析任務
報告者節點 (reporter): 生成最終研究報告

下面我們分別展開說下每個部分的亮點

前置分流（Coordinator）：智能體的“防火墻

作用： 攔截并直接處理無需啟動復雜流程的請求（如問候"你好"、能力詢問"你能干啥"、不當言論"你好討厭"）。
價值： 避免無關上下文污染核心流程，提升效率與用戶體驗。響應后終止對話，清空歷史消息（Discard Context策略）。

核心規劃模塊（Plan）:結構化推理與上下文處理

設計：
- 采用外層Plan + 內層Steps結構。Step定義了類型 (RESEARCH/PROCESSING)兩種智能體和每一步的詳細要求

class StepType(str, Enum):
    RESEARCH = "research"
    PROCESSING = "processing"


class Step(BaseModel):
    need_web_search: bool = Field(
        ..., description="Must be explicitly set for each step"
    )
    title: str
    description: str = Field(..., description="Specify exactly what data to collect")
    step_type: StepType = Field(..., description="Indicates the nature of the step")
    execution_res: Optional[str] = Field(
        default=None, description="The Step execution result"
    )


class Plan(BaseModel):
    locale: str = Field(
        ..., description="e.g. 'en-US' or 'zh-CN', based on the user's language"
    )
    has_enough_context: bool
    thought: str
    title: str
    steps: List[Step] = Field(
        default_factory=list,
        description="Research & Processing steps to get more context",
    )

子模塊：Plan里面還有兩個子模塊
- Human FeedBack：用戶后置判斷規劃是否合理
- Background_Investigator思考： 當前是用戶選擇是否開啟背景調研，但其實自動化執行更合理，類似RAG的多步思考，或者Pre-search策略，對于一些高時效性（超過模型訓練時間）的內容可以提高plan步驟的質量，但需警惕噪音的引入。
上下文處理：
- Background_Investigator: 僅基于用戶query搜索，搜索結果會處理成Human Message append到原有對話列表中。但會加入“background investigation results of user query:\n”的Prefix用于區分和原始用戶query的關系。
- Human_Feedback: 用戶反饋作為HumanMessage附加。
- Planner: 使用所有歷史Messages進行推理。

核心執行模塊（Research Team）：嚴格隔離與結果共享

負責執行Plan的就是Research Team節點，里面包含兩個智能體分別是擁有搜索和爬蟲工具的researcher和擁有python代碼工具的coder。這二者都使用了langgraph原生提供的react agent。

上下文處理：
- 上下文隔離 (Isolate Context)： Coder和Researcher的輸入僅包含當前Step的任務描述 (title, description)。彼此執行環境獨立，且與Planner隔離。
- 結構化輸入 (Reduce Noise)： 避免直接傳遞原始Message，而是將Step結構體轉換為清晰的任務指令模板，消除指代歧義（直接傳遞Message的問題就是在Prompt中很難指代什么是任務？）
```
agent_input = {
    "messages": [HumanMessage(content=f"#Task\n\n##title\n\n{step.title}\n\n##description\n\n{step.description}")]
}
```
- 結果共享： Coder/Researcher輸出保存在observations中，并作為HumanMessage更新全局Message列表（供Planner后續決策）。

這里的一個點是更新的Message列表會作為Plan的上文去進一步推理是否終止任務還是繼續收集加工信息，所以對于復雜任務，這一步其實給到Plan的壓力會比較大。原因有兩個

長度：如果前面規劃的step太多，這一步會有很多的message上文，長度超長，且內容多樣
復雜度：過多的內容給到plan模塊，模型特有的完美主義特色，往往讓模型總能找到更更多的優化點，去進一步開啟信息收集任務。

所以其實可以在輸出給Plan的Mesaage上文這一步進行信息壓縮處理，只返回任務完成狀態和一句話的信息收集概要。

最終報告撰寫(Reporter)

進入Reporter的條件： 最大循環次數、Plan規劃失敗、或Planner判定信息充足。
報告撰寫細節點
- 引用的處理：最終推理的上文是前面多個coder和researcher共同生成的，前面的步驟采用了在文本最后加入- [title](url)的markdown格式引用，最后推理也是相同。因此引用這里就沒有處理段落內部inline序號的問題了。但是對于過長的推理結果，文末引用并不太友好，但確實是最simple，native的解決方案
- 上文處理：項目使用了最原始的直接使用message列表，在不需要處理引用的情況下也okay
- 任務處理：和coder、reearcher相同reporter擁有獨立的任務上文也就是planner生成的最后推理總結任務。（但有個問題如果是超過最大循環次數或者失敗似乎沒看到任務生成的邏輯?）

框架總結

Deer-Flow是典型的線性、預定義角色、Supervisor-Worker架構（總-分-總模式）。其上下文管理核心在于嚴格隔離（Coordinator, Researcher, Coder, Background Investigator, Reporter 各自獨立）和 分階段結果傳遞（Reporter可見Researcher/Coder輸出）。有效運用了丟棄、隔離、壓縮等策略降低上下文負擔。

框架2：CoorAgent

Cooragnet

CoorAgent是清華推出的協作多智能體框架，在Deer-Flow基礎上，核心創新在于研究階段采用動態生成的智能體（Agent Factory模式），而非預定義的Researcher/Coder。其上下文管理策略與Deer-Flow類似，下文重點分析差異點。核心節點：

協調者 (Coordinator)： 同Deer-Flow（分流）。
規劃者 (Planner)： 同Deer-Flow（生成計劃），關鍵差異在于計劃中包含新智能體定義。
分發節點 (Publisher)： 分配待執行任務和智能體。
智能體工廠 (Agent Factory)： 根據Planner定義動態創建智能體。
智能體代理 (Agent Proxy)： 動態加載Agent配置，調用LangGraph預置的ReAct智能體執行任務。

核心規劃模塊（Planer）：任務規劃

設計： Planner不僅分解任務(Steps)，還直接定義執行該步驟所需的新智能體 (NewAgent)及其詳細配置（名稱name、角色role、能力capabilities、貢獻contribution）。Step包含指定agent_name、任務title/description和輸出要求note。
思考： 另一種設計是將智能體創建職責完全下放給Agent Factory，Planner僅專注于任務分解（目標、輸出格式、約束）。這能實現更好的節點解耦。

interface NewAgent {
  name: string;
  role: string;
  capabilities: string;
  contribution: string;
}

interface Step {
  agent_name: string;
  title: string;
  description: string;
  note?: string;
}

interface PlanWithAgents {
  thought: string;
  title: string;
  new_agents_needed: NewAgent[];
  steps: Step[];
}

分發節點（Publisher）：任務路由

個人感覺分發節點不是很必要，這一步只是判斷執行步驟以及是否終止，像Deer-Flow，OpenManus都采用了直接遍歷Plan步驟的方案，如果是需要判斷是否選擇終止，其實可以直接放到Plan的設計里面去。并且考慮新智能體之間的獨立無關性，應該可以走并發調用Agent Factory同時創建多智能體。

智能體工廠（Agent Factory）：動態創建

根據Planner要求使用模型直接設計新的智能體，包含智能體必備的幾大要素：任務描述、模型類型（文本、視覺），工具選擇，Prompt指令，如下結構體。

interface Tool {
  name: string;
  description: string;
}

interface AgentBuilder {
  agent_name: string;
  agent_description: string;
  thought: string;
  llm_type: string;
  selected_tools: Tool[];
  prompt: string;
}

Agent Factory的system prompt相對復雜，主要就是因為需要設計新智能體的System Prompt。所以其實這里可以借鑒Meta Prompt的思路，把構建Prompt的部分單獨拎出來，分兩步實現更干凈~

那這里簡單說下OpenAI推出的Meta-Prompting，簡單說就是openai從日常的prompt寫作中抽象了一些規則，并把這些規則總結成了Meta-Prompt（取Meta-Learning之義）,然后使用Meta-Prompt對你原始的簡單prompt進行細化，就能得到更為詳細的Prompt任務描述，如下


from openai import OpenAI

client = OpenAI()

META_PROMPT = """
Given a task description or existing prompt, produce a detailed system prompt to guide a language model in completing the task effectively.

# Guidelines

- Understand the Task: Grasp the main objective, goals, requirements, constraints, and expected output.
- Minimal Changes: If an existing prompt is provided, improve it only if it's simple. For complex prompts, enhance clarity and add missing elements without altering the original structure.
- Reasoning Before Conclusions**: Encourage reasoning steps before any conclusions are reached. ATTENTION! If the user provides examples where the reasoning happens afterward, REVERSE the order! NEVER START EXAMPLES WITH CONCLUSIONS!
    - Reasoning Order: Call out reasoning portions of the prompt and conclusion parts (specific fields by name). For each, determine the ORDER in which this is done, and whether it needs to be reversed.
    - Conclusion, classifications, or results should ALWAYS appear last.
- Examples: Include high-quality examples if helpful, using placeholders [in brackets] for complex elements.
   - What kinds of examples may need to be included, how many, and whether they are complex enough to benefit from placeholders.
- Clarity and Conciseness: Use clear, specific language. Avoid unnecessary instructions or bland statements.
- Formatting: Use markdown features for readability. DO NOT USE ``` CODE BLOCKS UNLESS SPECIFICALLY REQUESTED.
- Preserve User Content: If the input task or prompt includes extensive guidelines or examples, preserve them entirely, or as closely as possible. If they are vague, consider breaking down into sub-steps. Keep any details, guidelines, examples, variables, or placeholders provided by the user.
- Constants: DO include constants in the prompt, as they are not susceptible to prompt injection. Such as guides, rubrics, and examples.
- Output Format: Explicitly the most appropriate output format, in detail. This should include length and syntax (e.g. short sentence, paragraph, JSON, etc.)
    - For tasks outputting well-defined or structured data (classification, JSON, etc.) bias toward outputting a JSON.
    - JSON should never be wrapped in code blocks (```) unless explicitly requested.

The final prompt you output should adhere to the following structure below. Do not include any additional commentary, only output the completed system prompt. SPECIFICALLY, do not include any additional messages at the start or end of the prompt. (e.g. no "---")

[Concise instruction describing the task - this should be the first line in the prompt, no section header]

[Additional details as needed.]

[Optional sections with headings or bullet points for detailed steps.]

# Steps [optional]

[optional: a detailed breakdown of the steps necessary to accomplish the task]

# Output Format

[Specifically call out how the output should be formatted, be it response length, structure e.g. JSON, markdown, etc]

# Examples [optional]

[Optional: 1-3 well-defined examples with placeholders if necessary. Clearly mark where examples start and end, and what the input and output are. User placeholders as necessary.]
[If the examples are shorter than what a realistic example is expected to be, make a reference with () explaining how real examples should be longer / shorter / different. AND USE PLACEHOLDERS! ]

# Notes [optional]

[optional: edge cases, details, and an area to call or repeat out specific important considerations]
""".strip()

def generate_prompt(task_or_prompt: str):
    completion = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": META_PROMPT,
            },
            {
                "role": "user",
                "content": "Task, Goal, or Current Prompt:\n" + task_or_prompt,
            },
        ],
    )

    return completion.choices[0].message.content

前面Deer-Flow的博客中就提到它們項目的Prompt就是使用該方法由模型生成的。在我們的測試中Meta-Prompt也是有一些約束條件在

任務需要相對常見：人能簡單清洗描述任務目標、輸出格式、限制條件的比較適合。否則還是會有大量人的調優過程在
對模型有要求：不同能力的模型使用meta-prompt效果天差地別，qwen-plus對比Deepseek-v3，你會觀察到生成得到的prompt會顯著更短，更簡單。本質上Meta-Prompt通過對任務給出更多的要求、路徑、限制條件、備注來限制模型發揮，讓模型在任務上表現更穩定，那更強的模型對任務空間的劃分會更加細致。
生成模型和推理模型的一致性：使用Deepseek-v3得到的Prompt在qwen上表現并不好，但使用R1得到的Prompt在V3上表現很好，問題有可能來自是否同源、或者是否能力相似

技術實現：動態工作流編譯

CoorAgent因為存在動態生成agent，因此Langgraph實現會需要一些小巧思，因為graph編譯要求預定義好節點。 CoorAgent選擇在LangGraph外層自定義工作流引擎 (CompiledWorkflow)，管理節點間跳轉。如下


class CompiledWorkflow:
    def __init__(self, nodes: Dict[str, NodeFunc], edges: Dict[str, List[str]], start_node: str):
        self.nodes = nodes
        self.edges = edges
        self.start_node = start_node
        
    def invoke(self, state: State) -> State:
        current_node = self.start_node
        print(f"CompiledWorkflow current_node: {current_node}")
        while current_node != "__end__":
            if current_node not in self.nodes:
                raise ValueError(f"Node {current_node} not found in workflow")
                
            node_func = self.nodes[current_node]
            command = node_func(state)
            
            if hasattr(command, 'update') and command.update:
                for key, value in command.update.items():
                    print(f"update {key} to {value}")
                    state[key] = value
            
            current_node = command.goto
            
        return state

我們其實也碰到了類似動態智能體定義的問題，我們是選擇在node內部進行動態生成，這樣就繞開了動態節點的問題，還能更好利用langgraph其他branch，send之類的原生特性。

那多智能體就嘮這么多，后面該嘮嘮MCP了~

想看更全的大模型論文·微調預訓練數據·開源框架·AIGC應用 >> DecryPrompt

posted @ 2025-07-22 07:36 風雨中的小七閱讀(167) 評論(0) 收藏舉報

刷新頁面返回頂部