126 lines
4.2 KiB
Python
126 lines
4.2 KiB
Python
"""
|
||
封装 ADK Agent 调用
|
||
将 Dev Agent 的执行包装为可被任务管理器调用的异步函数
|
||
"""
|
||
import os
|
||
import sys
|
||
import asyncio
|
||
|
||
# 确保项目根目录在 path 里
|
||
PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
||
if PROJECT_ROOT not in sys.path:
|
||
sys.path.insert(0, PROJECT_ROOT)
|
||
|
||
from google.adk.runners import Runner
|
||
from google.adk.sessions import InMemorySessionService
|
||
from google.genai import types
|
||
|
||
|
||
class AgentRunner:
|
||
def __init__(self):
|
||
# 延迟导入 agent，避免循环导入
|
||
from my_agent.agent import root_agent
|
||
self.agent = root_agent
|
||
self.session_service = InMemorySessionService()
|
||
self._runner: Runner = None
|
||
|
||
def _get_runner(self) -> Runner:
|
||
if self._runner is None:
|
||
self._runner = Runner(
|
||
agent=self.agent,
|
||
app_name="dev_agent_server",
|
||
session_service=self.session_service,
|
||
auto_create_session=True,
|
||
)
|
||
return self._runner
|
||
|
||
async def run_task(self, task: dict) -> dict:
|
||
"""
|
||
执行一个开发任务，返回结构化结果
|
||
|
||
Args:
|
||
task: 任务字典，包含 description, project_path, requirements 等
|
||
|
||
Returns:
|
||
结构化的任务结果
|
||
"""
|
||
task_id = task["id"]
|
||
description = task["description"]
|
||
project_path = task.get("project_path", "")
|
||
requirements = task.get("requirements", "")
|
||
|
||
# 构建给 agent 的提示词
|
||
prompt = self._build_prompt(description, project_path, requirements)
|
||
|
||
print(f"[AgentRunner] 执行任务 {task_id}: {description[:60]}...")
|
||
|
||
runner = self._get_runner()
|
||
session_id = f"task_{task_id}"
|
||
user_id = "task_manager"
|
||
|
||
all_text = []
|
||
tool_calls = []
|
||
|
||
async for event in runner.run_async(
|
||
user_id=user_id,
|
||
session_id=session_id,
|
||
new_message=types.Content(
|
||
role="user",
|
||
parts=[types.Part(text=prompt)],
|
||
),
|
||
):
|
||
# 收集文本输出
|
||
if hasattr(event, 'output') and event.output:
|
||
content = event.output
|
||
if hasattr(content, 'parts'):
|
||
for part in content.parts:
|
||
if hasattr(part, 'text') and part.text:
|
||
all_text.append(part.text)
|
||
if hasattr(part, 'function_call') and part.function_call:
|
||
tool_calls.append({
|
||
"name": part.function_call.name,
|
||
"args": dict(part.function_call.args) if hasattr(part.function_call, 'args') else {},
|
||
})
|
||
|
||
result_text = "".join(all_text)
|
||
|
||
return {
|
||
"summary": self._extract_summary(result_text),
|
||
"full_response": result_text,
|
||
"tool_calls_count": len(tool_calls),
|
||
"tool_calls_sample": tool_calls[:10], # 只保留前 10 个
|
||
"status": "success" if result_text else "empty",
|
||
}
|
||
|
||
def _build_prompt(self, description: str, project_path: str, requirements: str) -> str:
|
||
"""构建给 agent 的任务指令"""
|
||
parts = [
|
||
"你需要完成以下开发任务：",
|
||
"",
|
||
f"**任务描述：**{description}",
|
||
]
|
||
if project_path:
|
||
parts.append(f"**项目路径：**{project_path}")
|
||
if requirements:
|
||
parts.append(f"**额外要求：**{requirements}")
|
||
parts.extend([
|
||
"",
|
||
"请按照你的工作流程执行：",
|
||
"1. 浏览项目结构，理解上下文",
|
||
"2. 编写或修改代码",
|
||
"3. 运行编译/构建验证",
|
||
"4. 完成后，按照你规定的报告格式输出结果",
|
||
"",
|
||
"请开始执行。",
|
||
])
|
||
return "\n".join(parts)
|
||
|
||
def _extract_summary(self, text: str) -> str:
|
||
"""从 agent 回复中提取摘要"""
|
||
# 简单处理：取前 1000 字符作为摘要
|
||
if not text:
|
||
return "(无响应)"
|
||
if len(text) <= 1000:
|
||
return text
|
||
return text[:1000] + "...（已截断）"
|