基于共享上下文和自主协作的 R&D Agent 生态系统 - qidao123.com技术社区-IT企服评测·应用市场

# Pseudo-code for MCP Service using FastAPI
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Dict, Any
import uuid
import time
app = FastAPI()
# In-memory storage for simplicity, replace with DB in production
contexts: Dict[str, Dict[str, Any]] = {}
class Message(BaseModel):
sender_id: str
sender_type: str # e.g., "agent", "human", "system"
type: str # e.g., "thought", "action", "observation", "tool_call", "tool_result", "human_input", "status_update"
content: Any
timestamp: float = Field(default_factory=time.time)
class TaskContext(BaseModel):
task_id: str
created_at: float = Field(default_factory=time.time)
messages: List[Message] = [] # History of interactions
shared_state: Dict[str, Any] = {} # Key-value pairs for shared state
# Add fields for linked knowledge, relevant documents etc.
@app.post("/tasks", response_model=TaskContext)
def create_task(initial_goal: str):
task_id = str(uuid.uuid4())
new_context = TaskContext(task_id=task_id, shared_state={"status": "CREATED", "initial_goal": initial_goal})
contexts[task_id] = new_context.model_dump() # Store as dict for mutation
print(f"Task {task_id} created.")
return new_context
@app.get("/tasks/{task_id}/context", response_model=TaskContext)
def get_context(task_id: str):
if task_id not in contexts:
raise HTTPException(status_code=404, detail="Task not found")
return TaskContext(**contexts[task_id]) # Return as Pydantic model
@app.post("/tasks/{task_id}/messages")
def add_message(task_id: str, message: Message):
if task_id not in contexts:
raise HTTPException(status_code=404, detail="Task not found")
contexts[task_id]['messages'].append(message.model_dump())
# Optional: Trigger event notification here
print(f"Task {task_id}: Added message from {message.sender_id} ({message.type})")
return {"status": "success"}
@app.put("/tasks/{task_id}/state")
def update_shared_state(task_id: str, state_update: Dict[str, Any]):
if task_id not in contexts:
raise HTTPException(status_code=404, detail="Task not found")
contexts[task_id]['shared_state'].update(state_update)
# Optional: Trigger event notification if specific keys updated
print(f"Task {task_id}: Updated shared state: {state_update}")
return {"status": "success"}
@app.get("/tasks/{task_id}/state", response_model=Dict[str, Any])
def get_shared_state(task_id: str):
if task_id not in contexts:
raise HTTPException(status_code=404, detail="Task not found")
return contexts[task_id]['shared_state']
# Run with: uvicorn mcp_service:app --reload

复制代码

# Pseudo-code for a Literature Review Agent's internal logic (Simplified)
from agent_core_base import AgentCore # Base class handles LLM, Memory, Tools
from tools import SearchPubMedTool, SummarizeTextTool
from mcp_client import MCPClient # Custom client for MCP Service API
class LiteratureReviewAgent(AgentCore): # Inherit from base Agent class
def __init__(self, id, llm, memory, tools, mcp_client: MCPClient, task_id: str):
super().__init__(id, llm, memory, tools)
self.mcp_client = mcp_client
self.task_id = task_id # Agent is initialized for a specific task context
def run(self):
# Agent starts its execution loop
print(f"Agent {self.id}: Starting execution for task {self.task_id}")
try:
while True:
# 1. Get current context from MCP
context = self.mcp_client.get_context(self.task_id)
current_history = context['messages']
shared_state = context['shared_state']
# Check if task is already completed or requires human intervention
if shared_state.get('status') == 'COMPLETED' or shared_state.get('status') == 'NEEDS_HUMAN_REVIEW':
print(f"Agent {self.id}: Task status is {shared_state.get('status')}. Exiting.")
break # Exit loop
# 2. LLM Reasoning based on context
# Craft a prompt that includes the history and state from context
prompt = self.generate_reasoning_prompt(current_history, shared_state)
llm_response = self.llm.chat(prompt) # Assuming chat model
thought = extract_thought_from_llm_response(llm_response) # Needs parsing
action = extract_action_from_llm_response(llm_response) # Needs parsing (tool call or update_state)
# 3. Log thought to MCP
self.mcp_client.add_message(self.task_id, self.id, "thought", thought)
if action:
action_type = action['type']
action_details = action['details']
# 4. Log action to MCP
self.mcp_client.add_message(self.task_id, self.id, "action", action)
if action_type == "tool_call":
tool_name = action_details['tool_name']
tool_params = action_details['tool_params']
print(f"Agent {self.id}: Calling tool {tool_name} with {tool_params}")
try:
tool_result = self.use_tool(tool_name, tool_params)
print(f"Agent {self.id}: Tool {tool_name} returned {tool_result}")
# 5. Log observation/tool result to MCP
self.mcp_client.add_message(self.task_id, self.id, "observation", tool_result)
self.mcp_client.add_message(self.task_id, self.id, "tool_result", tool_result)
# 6. Based on result, maybe update shared state?
if self.check_if_literature_complete(tool_result): # Agent's own logic
self.mcp_client.update_shared_state(self.task_id, {"literature_review_status": "COMPLETED", "literature_summary": tool_result['summary']})
print(f"Agent {self.id}: Literature review task completed.")
# Agent might choose to exit here, or LLM might decide the next step
# break # Example: Exit after completion
except Exception as tool_error:
error_msg = f"Tool {tool_name} failed: {tool_error}"
print(error_msg)
# 5. Log error to MCP
self.mcp_client.add_message(self.task_id, self.id, "observation", {"error": error_msg})
self.mcp_client.update_shared_state(self.task_id, {"status": "ERROR_IN_LIT_REVIEW", "error_details": error_msg})
# Agent might decide to stop or seek help
break # Example: Exit on major error
elif action_type == "update_state":
state_updates = action_details['updates']
self.mcp_client.update_shared_state(self.task_id, state_updates)
print(f"Agent {self.id}: Updated shared state with {state_updates}")
elif action_type == "delegate": # Example: Agent decides to delegate to another agent type
target_agent_type = action_details['agent_type']
delegation_task = action_details['task_description']
# Log this delegation to the context so other agents can see it
self.mcp_client.add_message(self.task_id, self.id, "delegation", {"to_agent_type": target_agent_type, "description": delegation_task})
# The Session Manager or another Agent might pick this up and launch the new agent
# This requires other components monitoring the context
elif action_type == "report_completion": # Agent signals its part is done
final_summary = action_details['summary']
self.mcp_client.update_shared_state(self.task_id, {"literature_review_status": "FINALIZED", "final_summary": final_summary})
self.mcp_client.add_message(self.task_id, self.id, "status_update", {"status": "Literature review finalized"})
print(f"Agent {self.id}: Reported finalization.")
# break # Agent might exit after its specific sub-task is done
# ... other action types (e.g., ask_human_for_help)
else: # LLM didn't generate a valid action or signaled completion implicitly
# Needs logic to detect if LLM is stuck or implicitly finished
print(f"Agent {self.id}: LLM returned no valid action or finished.")
# Maybe update state to signal potential stagnation or sub-task completion?
# self.mcp_client.update_shared_state(self.task_id, {"status": "LIT_REVIEW_AGENT_STALLED"})
break # Example: Exit loop
time.sleep(5) # Prevent tight loop, allow other agents to potentially act
except Exception as e:
print(f"Agent {self.id} encountered fatal error: {e}")
# Log fatal error and update global state
self.mcp_client.add_message(self.task_id, self.id, "system_error", {"error": str(e)})
self.mcp_client.update_shared_state(self.task_id, {"status": "FATAL_ERROR", "error_source": self.id, "error_details": str(e)})
def generate_reasoning_prompt(self, history, shared_state):
# Craft the core prompt for the LLM, injecting relevant history and state
history_text = "\n".join([f"{msg['sender_type']} ({msg['sender_id']}) [{msg['type']}]: {msg['content']}" for msg in history])
state_text = json.dumps(shared_state, indent=2)
prompt = f"""
You are a {self.agent_type} with ID {self.id}. Your current task context (ID: {self.task_id}) is managed by the Master Context Protocol Service.
Review the task history and shared state below. Determine your next step to contribute to the overall R&D goal defined in the shared state ('initial_goal').
Your actions should be one of the following types: 'tool_call', 'update_state', 'delegate', 'report_completion', 'ask_human_for_help'.
Output your thought process first, then your chosen action in JSON format. If no action is needed now, explain why.
Current Shared State:
{state_text}
Task History:
{history_text}
Available Tools: {list(self.tools.keys())}
Think step-by-step. What is the current situation? What needs to be done next based on the goal and state? Which action is most appropriate?
"""
return prompt
def check_if_literature_complete(self, tool_result):
# Agent-specific logic to determine if its sub-task is done
# E.g., check if a summary is generated, if enough papers were reviewed, etc.
pass # Placeholder
# use_tool method would be similar to previous examples, calling external services
# class AgentCoreBase would handle the LLM interaction and basic structure

复制代码

# Pseudo-code for Task Initiator (e.g., a simple script or service)
from mcp_client import MCPClient
mcp_client = MCPClient(mcp_service_url="http://mcp_service:8000") # Point to MCP Service
def start_new_rnd_task(user_goal: str):
print(f"Initiating R&D task with goal: {user_goal}")
# 1. Create a new context for the task
new_context = mcp_client.create_task(user_goal)
task_id = new_context['task_id']
print(f"New task created with ID: {task_id}")
# 2. Start initial Agent(s) and provide them with the task_id
# This requires knowing which agent(s) should start based on the initial goal
# This logic could be hardcoded, rule-based, or even LLM-decided by the Initiator.
initial_agent_types = ["Literature Review Agent", "Hypothesis Generation Agent"] # Example
for agent_type in initial_agent_types:
# Assume a way to launch an agent instance (e.g., calling a Kubernetes API to create a Pod,
# or calling an Agent Management Service)
launch_agent_instance(agent_type, task_id) # This function starts the Agent process
print(f"Launched initial agent: {agent_type} for task {task_id}")
print("Initial agents launched. Monitoring task state via MCP.")
# Pseudo-code for Global State Monitor (e.g., a background process or UI component)
def monitor_rnd_tasks():
mcp_client = MCPClient(mcp_service_url="http://mcp_service:8000")
while True:
# 1. Get overview of tasks (e.g., list all tasks or recent ones)
tasks_overview = mcp_client.list_tasks() # Needs a list_tasks endpoint in MCP Service
for task_summary in tasks_overview:
task_id = task_summary['task_id']
current_state = mcp_client.get_shared_state(task_id)
print(f"Task {task_id}: Status - {current_state.get('status', 'N/A')}, Goal - {current_state.get('initial_goal', 'N/A')}")
# Check for specific conditions needing human attention
if current_state.get('status') == 'NEEDS_HUMAN_REVIEW':
print(f"ALERT: Task {task_id} needs human review!")
# Trigger notification (email, slack) and provide link to UI showing context
if current_state.get('status') == 'FATAL_ERROR':
print(f"ALERT: Task {task_id} encountered FATAL ERROR!")
# Trigger notification
time.sleep(60) # Check every minute
# launch_agent_instance(agent_type, task_id) pseudo-code:
# This function is part of the Agent Management layer, not MCP itself.
# It would typically interact with Kubernetes to deploy a pod,
# passing task_id and MCP Service endpoint as environment variables or arguments
# so the Agent knows which context to join.
def launch_agent_instance(agent_type, task_id):
# Example: Call Kubernetes API to create a Pod
k8s_client = get_kubernetes_client()
pod_manifest = {
"apiVersion": "v1",
"kind": "Pod",
"metadata": {"generateName": f"{agent_type.lower().replace(' ', '-')}-"},
"spec": {
"containers": [{
"name": "agent",
"image": f"your_agent_image:{agent_type.lower().replace(' ', '-')}", # Need different images or config for each type
"env": [
{"name": "AGENT_ID", "value": str(uuid.uuid4())},
{"name": "AGENT_TYPE", "value": agent_type},
{"name": "TASK_ID", "value": task_id},
{"name": "MCP_SERVICE_URL", "value": "http://mcp_service:8000"},
# ... other env vars for LLM endpoint, DBs, etc.
]
}],
"restartPolicy": "Never" # Or OnFailure depending on desired behavior
}
}
k8s_client.create_namespaced_pod("default", pod_manifest)
print(f"Requested Kubernetes to launch {agent_type} pod for task {task_id}")

复制代码