forked from zhanggangyong/agent_management
2069 lines
70 KiB
Python
2069 lines
70 KiB
Python
"""
|
||
Agent 代码生成模块
|
||
使用 AI 根据工具定义生成 Pydantic Agent 代码
|
||
基于 agent_templates/agents/_template 框架
|
||
包含完整 CI/CD 配置
|
||
"""
|
||
|
||
import os
|
||
import json
|
||
import logging
|
||
import re
|
||
import requests
|
||
from typing import Dict, List, Optional, Any
|
||
from datetime import datetime
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class AgentCodeGenerator:
|
||
"""Agent 代码生成器 - 基于 _template 框架"""
|
||
|
||
def __init__(self):
|
||
# LLM 配置 (用于 AI 生成代码)
|
||
self.llm_base_url = os.getenv(
|
||
"LLM_BASE_URL",
|
||
"https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io/v1"
|
||
)
|
||
self.llm_api_key = os.getenv("LLM_API_KEY", "sk-mPV5MVVVVvfGSkXA-ASQXQ")
|
||
self.llm_model = os.getenv("LLM_MODEL", "taiji/claude-sonnet-4-5")
|
||
|
||
# ACR 配置
|
||
self.acr_registry = os.getenv("ACR_REGISTRY", "agnettaiji.azurecr.io")
|
||
self.acr_namespace = os.getenv("ACR_NAMESPACE", "ai-agents")
|
||
|
||
# CI/CD Secrets 默认值（用于文档说明）
|
||
self.cicd_secrets = {
|
||
"ACR_LOGIN_SERVER": "agnettaiji.azurecr.io",
|
||
"ACR_USERNAME": "agnettaiji",
|
||
"ACR_PASSWORD": "hDpX5t34N5ZmnKdtqyjYL5co/SnXJrmD20CRpGpWaG+ACRCw2wGM",
|
||
"AZ_CLIENT_ID": "fb306798-2cfe-4ac9-ba48-eab7bc71bcfe",
|
||
"AZ_CLIENT_SECRET": "cVK8Q~xlfBwm2_t2TC24yrTukWV4F3G~eIjBBa0D",
|
||
"AZ_TENANT_ID": "263c3ff6-1be5-4141-8308-b188464fb297",
|
||
"AZ_SUBSCRIPTION_ID": "c6c47e4c-f5f4-49f8-b26f-7728862c17d6",
|
||
"AZ_RG": "taiji-ai-pda",
|
||
"AZ_AKS": "taiji-ai-pda",
|
||
"AZURE_DNS_ZONE": "taijiagnet.com"
|
||
}
|
||
|
||
def _call_llm(self, prompt: str, system_prompt: str = None) -> str:
|
||
"""调用 LLM 生成代码"""
|
||
messages = []
|
||
|
||
if system_prompt:
|
||
messages.append({"role": "system", "content": system_prompt})
|
||
|
||
messages.append({"role": "user", "content": prompt})
|
||
|
||
try:
|
||
response = requests.post(
|
||
f"{self.llm_base_url}/chat/completions",
|
||
headers={
|
||
"Authorization": f"Bearer {self.llm_api_key}",
|
||
"Content-Type": "application/json"
|
||
},
|
||
json={
|
||
"model": self.llm_model,
|
||
"messages": messages,
|
||
"temperature": 0.2,
|
||
"max_tokens": 8000
|
||
},
|
||
timeout=120
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
return result["choices"][0]["message"]["content"]
|
||
else:
|
||
logger.error(f"LLM 调用失败: {response.text}")
|
||
return None
|
||
|
||
except Exception as e:
|
||
logger.error(f"LLM 调用异常: {e}")
|
||
return None
|
||
|
||
def _convert_name_to_python(self, name: str) -> str:
|
||
"""将工具名称转换为 Python 函数名"""
|
||
name = re.sub(r'[^a-zA-Z0-9_]', '_', name)
|
||
if name and name[0].isdigit():
|
||
name = '_' + name
|
||
return name.lower()
|
||
|
||
def _json_type_to_python(self, json_type: str) -> str:
|
||
"""JSON Schema 类型转 Python 类型"""
|
||
type_map = {
|
||
"string": "str",
|
||
"integer": "int",
|
||
"number": "float",
|
||
"boolean": "bool",
|
||
"array": "list",
|
||
"object": "dict"
|
||
}
|
||
return type_map.get(json_type, "Any")
|
||
|
||
async def generate_tool_code_with_ai(self, tool_config: dict, api_key: str = None) -> str:
|
||
"""
|
||
使用 AI 理解工具定义并生成代码（更智能，支持复杂场景如 URL 拼接）
|
||
|
||
Args:
|
||
tool_config: 工具配置
|
||
api_key: LLM API Key
|
||
|
||
Returns:
|
||
AI 生成的 Python 代码
|
||
"""
|
||
import httpx
|
||
|
||
# 使用实例配置或环境变量（默认使用 claude-sonnet 生成高质量代码）
|
||
llm_url = self.llm_base_url
|
||
llm_model = self.llm_model # 默认: taiji/claude-sonnet-4-5
|
||
llm_key = api_key or self.llm_api_key
|
||
|
||
if not llm_key:
|
||
# 如果没有 API Key，回退到模板生成
|
||
logger.warning("未提供 LLM API Key，使用模板生成")
|
||
return self.generate_tool_code(tool_config)
|
||
|
||
name = tool_config.get("name", "custom_tool")
|
||
func_name = self._convert_name_to_python(name)
|
||
|
||
# 构建 prompt
|
||
prompt = f"""你是一个 Python 代码生成专家。根据以下工具定义，生成一个异步 Python 函数。
|
||
|
||
## 工具定义
|
||
- 名称: {tool_config.get("name")}
|
||
- 描述: {tool_config.get("description")}
|
||
- API URL: {tool_config.get("url")}
|
||
- HTTP 方法: {tool_config.get("method", "GET")}
|
||
- 认证方式: {json.dumps(tool_config.get("auth", {}), ensure_ascii=False)}
|
||
- 请求参数: {json.dumps(tool_config.get("request_params", tool_config.get("input_schema", {})), ensure_ascii=False)}
|
||
- 请求体: {json.dumps(tool_config.get("request_body", {}), ensure_ascii=False)}
|
||
|
||
## 重要提示
|
||
1. 如果 URL 像 `https://r.jina.ai/` 这样需要将参数拼接到路径中（如 `https://r.jina.ai/{{target_url}}`），请正确处理 URL 拼接
|
||
2. **认证方式处理**：
|
||
- 如果认证类型是 `bearer`，必须使用 `Authorization: Bearer <token>` 格式的 header（不是 X-API-Key！）
|
||
- 从环境变量 `TOOL_API_KEY` 获取 token：`os.getenv("TOOL_API_KEY")`
|
||
- 示例：`headers = {{"Authorization": f"Bearer {{os.getenv('TOOL_API_KEY')}}"}}`
|
||
3. 函数必须是 async def，返回 JSON 字符串
|
||
4. 使用 httpx 作为 HTTP 客户端
|
||
5. 包含完整的错误处理
|
||
|
||
## 输出格式
|
||
只输出 Python 代码，不要其他解释。代码格式如下：
|
||
|
||
```python
|
||
\"\"\"
|
||
工具: {{name}}
|
||
描述: {{description}}
|
||
\"\"\"
|
||
import os
|
||
import json
|
||
from typing import Optional, Any
|
||
import httpx
|
||
|
||
async def {func_name}(...) -> str:
|
||
...
|
||
```"""
|
||
|
||
try:
|
||
async with httpx.AsyncClient(timeout=60.0) as client:
|
||
response = await client.post(
|
||
f"{llm_url}/chat/completions",
|
||
headers={
|
||
"Authorization": f"Bearer {llm_key}",
|
||
"Content-Type": "application/json"
|
||
},
|
||
json={
|
||
"model": llm_model,
|
||
"messages": [
|
||
{"role": "system", "content": "你是一个专业的 Python 代码生成器，只输出代码，不要解释。"},
|
||
{"role": "user", "content": prompt}
|
||
],
|
||
"temperature": 0.2,
|
||
"max_tokens": 2000
|
||
}
|
||
)
|
||
response.raise_for_status()
|
||
|
||
content = response.json()["choices"][0]["message"]["content"]
|
||
|
||
# 提取代码块
|
||
if "```python" in content:
|
||
code = content.split("```python")[1].split("```")[0].strip()
|
||
elif "```" in content:
|
||
code = content.split("```")[1].split("```")[0].strip()
|
||
else:
|
||
code = content.strip()
|
||
|
||
logger.info(f"✅ AI 成功生成工具代码: {name}")
|
||
return code
|
||
|
||
except Exception as e:
|
||
logger.error(f"AI 生成代码失败: {e}，回退到模板生成")
|
||
return self.generate_tool_code(tool_config)
|
||
|
||
def generate_tool_code(self, tool_config: dict) -> str:
|
||
"""
|
||
根据工具配置生成 Pydantic 工具代码（模板方式）
|
||
|
||
Args:
|
||
tool_config: 包含 name, url, method, auth, request_params/input_schema 等
|
||
|
||
Returns:
|
||
生成的 Python 代码字符串
|
||
"""
|
||
name = tool_config.get("name", "custom_tool")
|
||
func_name = self._convert_name_to_python(name)
|
||
desc = tool_config.get("description", "")
|
||
url = tool_config.get("url", "")
|
||
method = tool_config.get("method", "GET").upper()
|
||
auth = tool_config.get("auth", {})
|
||
# 兼容 request_params 和 input_schema 两种字段名
|
||
request_params = tool_config.get("request_params") or tool_config.get("input_schema") or {}
|
||
request_body = tool_config.get("request_body", {})
|
||
timeout = tool_config.get("timeout", 30)
|
||
|
||
# 构建参数
|
||
params = []
|
||
params_doc = []
|
||
required_params = []
|
||
|
||
# 支持两种格式：
|
||
# 1. {"properties": {"symbol": {...}}, "required": ["symbol"]}
|
||
# 2. {"symbol": {...}} (直接参数格式)
|
||
param_props = request_params
|
||
if request_params and request_params.get("properties"):
|
||
param_props = request_params["properties"]
|
||
required_params = request_params.get("required", [])
|
||
elif request_params and not any(k in request_params for k in ["type", "required", "description"]):
|
||
param_props = request_params
|
||
else:
|
||
param_props = {}
|
||
|
||
if param_props:
|
||
for p_name, p_info in param_props.items():
|
||
if not isinstance(p_info, dict):
|
||
continue
|
||
p_type = self._json_type_to_python(p_info.get("type", "string"))
|
||
p_desc = p_info.get("description", "")
|
||
# 检查是否在 required 列表中
|
||
is_required = p_name in required_params or p_info.get("required", False)
|
||
default = p_info.get("default")
|
||
|
||
if is_required:
|
||
params.append(f"{p_name}: {p_type}")
|
||
else:
|
||
default_val = f'"{default}"' if isinstance(default, str) else (default if default is not None else "None")
|
||
params.append(f"{p_name}: Optional[{p_type}] = {default_val}")
|
||
|
||
params_doc.append(f" {p_name}: {p_desc}")
|
||
|
||
params_str = ", ".join(params) if params else ""
|
||
params_doc_str = "\n".join(params_doc) if params_doc else " 无参数"
|
||
|
||
# 生成认证代码
|
||
auth_headers = self._get_auth_headers_code(auth)
|
||
|
||
# 构建参数字典代码
|
||
params_dict_code = ""
|
||
if param_props:
|
||
params_dict_items = []
|
||
for p_name in param_props.keys():
|
||
if isinstance(param_props[p_name], dict):
|
||
params_dict_items.append(f'"{p_name}": {p_name}')
|
||
if params_dict_items:
|
||
params_dict_code = "params = {" + ", ".join(params_dict_items) + "}"
|
||
else:
|
||
params_dict_code = "params = {}"
|
||
else:
|
||
params_dict_code = "params = {}"
|
||
|
||
# API Key in query
|
||
if auth and auth.get("type") == "api_key" and auth.get("in") == "query":
|
||
key_name = auth.get("name", "apikey")
|
||
params_dict_code += f'\n params["{key_name}"] = os.getenv("TOOL_API_KEY", "")'
|
||
|
||
# 生成 URL 代码（使用 api_url 避免与参数名冲突）
|
||
url_code = f'api_url = "{url}"'
|
||
|
||
# 根据 HTTP 方法决定参数传递方式
|
||
# POST/PUT/PATCH: 参数放到请求体 (json)
|
||
# GET/DELETE: 参数放到查询参数 (params)
|
||
if method in ["POST", "PUT", "PATCH"]:
|
||
# POST 请求：参数作为 JSON 请求体
|
||
request_code = f'''async with httpx.AsyncClient(timeout={timeout}) as client:
|
||
response = await client.request(
|
||
method="{method}",
|
||
url=api_url,
|
||
headers=headers,
|
||
json={{k: v for k, v in params.items() if v is not None}}
|
||
)'''
|
||
else:
|
||
# GET 请求：参数作为查询参数
|
||
request_code = f'''async with httpx.AsyncClient(timeout={timeout}) as client:
|
||
response = await client.request(
|
||
method="{method}",
|
||
url=api_url,
|
||
headers=headers,
|
||
params={{k: v for k, v in params.items() if v is not None}}
|
||
)'''
|
||
|
||
# 生成函数代码
|
||
code = f'''"""
|
||
工具: {name}
|
||
描述: {desc}
|
||
自动生成时间: {datetime.now().isoformat()}
|
||
"""
|
||
import os
|
||
import json
|
||
from typing import Optional, Any
|
||
import httpx
|
||
|
||
|
||
async def {func_name}({params_str}) -> str:
|
||
"""
|
||
{desc}
|
||
|
||
Args:
|
||
{params_doc_str}
|
||
|
||
Returns:
|
||
API 响应结果 (JSON 格式)
|
||
"""
|
||
{url_code}
|
||
{auth_headers}
|
||
{params_dict_code}
|
||
|
||
try:
|
||
{request_code}
|
||
|
||
if response.status_code == 200:
|
||
return json.dumps({{
|
||
"success": True,
|
||
"data": response.json() if response.headers.get("content-type", "").startswith("application/json") else response.text
|
||
}}, ensure_ascii=False, indent=2)
|
||
else:
|
||
return json.dumps({{
|
||
"success": False,
|
||
"status_code": response.status_code,
|
||
"error": response.text[:500]
|
||
}}, ensure_ascii=False)
|
||
|
||
except Exception as e:
|
||
return json.dumps({{
|
||
"success": False,
|
||
"error": str(e)
|
||
}}, ensure_ascii=False)
|
||
'''
|
||
return code
|
||
|
||
def generate_mcp_server(
|
||
self,
|
||
agent_name: str,
|
||
description: str,
|
||
tools_config: List[Dict]
|
||
) -> str:
|
||
"""
|
||
生成 MCP 服务器代码 (mcp_server.py)
|
||
基于 _template/src/server/mcp_server.py 框架
|
||
"""
|
||
# 生成工具函数
|
||
tool_functions = []
|
||
tool_map_entries = []
|
||
tool_list_entries = []
|
||
|
||
for tool in tools_config:
|
||
name = tool.get("name", "custom_tool")
|
||
func_name = self._convert_name_to_python(name)
|
||
desc = tool.get("description", "")
|
||
url = tool.get("url", "")
|
||
method = tool.get("method", "GET").upper()
|
||
auth = tool.get("auth", {})
|
||
request_params = tool.get("request_params", {})
|
||
timeout = tool.get("timeout", 30)
|
||
|
||
# 构建参数
|
||
params = []
|
||
params_doc = []
|
||
properties = {}
|
||
required_params = []
|
||
|
||
# 支持两种格式：
|
||
# 1. {"properties": {"symbol": {...}}}
|
||
# 2. {"symbol": {...}} (直接参数格式)
|
||
param_props = request_params
|
||
if request_params and request_params.get("properties"):
|
||
param_props = request_params["properties"]
|
||
elif request_params and not any(k in request_params for k in ["type", "required", "description"]):
|
||
# 直接参数格式
|
||
param_props = request_params
|
||
else:
|
||
param_props = {}
|
||
|
||
if param_props:
|
||
for p_name, p_info in param_props.items():
|
||
if not isinstance(p_info, dict):
|
||
continue
|
||
p_type = self._json_type_to_python(p_info.get("type", "string"))
|
||
p_desc = p_info.get("description", "")
|
||
is_required = p_info.get("required", False)
|
||
default = p_info.get("default")
|
||
|
||
if is_required:
|
||
params.append(f"{p_name}: {p_type}")
|
||
required_params.append(p_name)
|
||
else:
|
||
default_val = f'"{default}"' if isinstance(default, str) else (default if default is not None else "None")
|
||
params.append(f"{p_name}: Optional[{p_type}] = {default_val}")
|
||
|
||
params_doc.append(f" {p_name}: {p_desc}")
|
||
properties[p_name] = {"type": p_info.get("type", "string"), "description": p_desc}
|
||
|
||
params_str = ", ".join(params) if params else ""
|
||
params_doc_str = "\n".join(params_doc) if params_doc else " 无参数"
|
||
|
||
# 生成认证代码
|
||
auth_headers = self._get_auth_headers_code(auth)
|
||
|
||
# 构建参数字典代码
|
||
params_dict_code = ""
|
||
if param_props:
|
||
params_dict_code = "params = {"
|
||
for p_name in param_props.keys():
|
||
if isinstance(param_props[p_name], dict):
|
||
params_dict_code += f'"{p_name}": {p_name}, '
|
||
params_dict_code = params_dict_code.rstrip(", ") + "}"
|
||
else:
|
||
params_dict_code = "params = {}"
|
||
|
||
# API Key in query
|
||
if auth and auth.get("type") == "api_key" and auth.get("in") == "query":
|
||
key_name = auth.get("name", "apikey")
|
||
params_dict_code += f'\n params["{key_name}"] = os.getenv("TOOL_API_KEY", "")'
|
||
|
||
# 生成函数代码
|
||
func_code = f'''
|
||
@server.tool()
|
||
async def {func_name}({params_str}) -> str:
|
||
"""
|
||
{desc}
|
||
|
||
Args:
|
||
{params_doc_str}
|
||
|
||
Returns:
|
||
API 响应结果 (JSON 格式)
|
||
"""
|
||
import httpx
|
||
|
||
url = "{url}"
|
||
{auth_headers}
|
||
{params_dict_code}
|
||
|
||
try:
|
||
async with httpx.AsyncClient(timeout={timeout}) as client:
|
||
response = await client.request(
|
||
method="{method}",
|
||
url=url,
|
||
headers=headers,
|
||
params={{k: v for k, v in params.items() if v is not None}}
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
return json.dumps({{
|
||
"success": True,
|
||
"data": response.json() if response.headers.get("content-type", "").startswith("application/json") else response.text
|
||
}}, ensure_ascii=False, indent=2)
|
||
else:
|
||
return json.dumps({{
|
||
"success": False,
|
||
"status_code": response.status_code,
|
||
"error": response.text[:500]
|
||
}}, ensure_ascii=False)
|
||
|
||
except Exception as e:
|
||
# 使用 AI Agent 作为后备
|
||
result = await get_agent().run(f"请帮我处理这个请求: {params}")
|
||
return json.dumps({{
|
||
"success": True,
|
||
"source": "ai_agent",
|
||
"result": result.output
|
||
}}, ensure_ascii=False, indent=2)
|
||
'''
|
||
tool_functions.append(func_code)
|
||
tool_map_entries.append(f" '{func_name}': {func_name},")
|
||
|
||
tool_list_entries.append(f''' {{
|
||
"name": "{func_name}",
|
||
"description": "{desc}",
|
||
"inputSchema": {{
|
||
"type": "object",
|
||
"properties": {json.dumps(properties, ensure_ascii=False)},
|
||
"required": {json.dumps(required_params)}
|
||
}}
|
||
}},''')
|
||
|
||
tool_functions_str = "\n".join(tool_functions)
|
||
tool_map_str = "\n".join(tool_map_entries)
|
||
tool_list_str = "\n".join(tool_list_entries)
|
||
|
||
# 生成系统提示词
|
||
system_prompt = self._generate_system_prompt(agent_name, description, tools_config)
|
||
|
||
return f'''"""
|
||
MCP 服务器 - {agent_name}
|
||
|
||
{description}
|
||
使用 Pydantic AI 和 FastMCP 框架。
|
||
自动生成时间: {datetime.now().isoformat()}
|
||
"""
|
||
import json
|
||
import os
|
||
from typing import Optional
|
||
|
||
from mcp.server.fastmcp import FastMCP
|
||
from pydantic_ai import Agent
|
||
|
||
# ==================== 配置 ====================
|
||
|
||
# LiteLLM Gateway 配置
|
||
_BASE_URL = os.getenv('OPENAI_BASE_URL',
|
||
os.getenv('LLM_BASE_URL', 'https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io/v1'))
|
||
_API_KEY = os.getenv('OPENAI_API_KEY', 'sk')
|
||
|
||
os.environ.setdefault('OPENAI_API_KEY', _API_KEY)
|
||
os.environ.setdefault('OPENAI_BASE_URL', _BASE_URL)
|
||
|
||
# 模型名称（pydantic_ai 需要 openai: 前缀）
|
||
def _get_model_name() -> str:
|
||
model = os.getenv('MODEL_NAME', os.getenv('LITELLM_MODEL', 'taiji/gpt-4o-mini'))
|
||
return model if ':' in model else f'openai:{{model}}'
|
||
|
||
MODEL_NAME = _get_model_name()
|
||
|
||
# ==================== MCP 服务器 ====================
|
||
|
||
server = FastMCP('{agent_name}')
|
||
|
||
# 系统提示词
|
||
SYSTEM_PROMPT = """{system_prompt}"""
|
||
|
||
|
||
def get_agent() -> Agent:
|
||
"""创建 Agent 实例（每次调用使用最新的 API Key）"""
|
||
return Agent(MODEL_NAME, system_prompt=SYSTEM_PROMPT)
|
||
|
||
|
||
# ==================== MCP 工具定义 ====================
|
||
{tool_functions_str}
|
||
|
||
# ==================== 工具映射（供 API 使用）====================
|
||
|
||
TOOL_MAP = {{
|
||
{tool_map_str}
|
||
}}
|
||
|
||
TOOL_LIST = [
|
||
{tool_list_str}
|
||
]
|
||
|
||
|
||
if __name__ == '__main__':
|
||
server.run()
|
||
'''
|
||
|
||
def _get_auth_headers_code(self, auth: Dict) -> str:
|
||
"""生成认证头代码"""
|
||
if not auth:
|
||
return "headers = {}"
|
||
|
||
auth_type = auth.get("type", "")
|
||
|
||
if auth_type == "api_key":
|
||
key = auth.get("key", "")
|
||
location = auth.get("in", "header")
|
||
name = auth.get("name", "X-API-Key")
|
||
|
||
if location == "header":
|
||
return f'headers = {{"{name}": os.getenv("TOOL_API_KEY", "{key}")}}'
|
||
else:
|
||
return "headers = {} # API Key 在查询参数中"
|
||
|
||
elif auth_type == "bearer":
|
||
key = auth.get("key", "")
|
||
return f'headers = {{"Authorization": f"Bearer {{os.getenv(\'TOOL_API_KEY\', \'{key}\')}}" }}'
|
||
|
||
elif auth_type == "basic":
|
||
username = auth.get("username", "")
|
||
password = auth.get("password", "")
|
||
return f'''import base64
|
||
_creds = base64.b64encode(f"{{os.getenv('TOOL_USERNAME', '{username}')}}:{{os.getenv('TOOL_PASSWORD', '{password}')}}".encode()).decode()
|
||
headers = {{"Authorization": f"Basic {{_creds}}"}}'''
|
||
|
||
return "headers = {}"
|
||
|
||
def _generate_system_prompt(
|
||
self,
|
||
agent_name: str,
|
||
description: str,
|
||
tools_config: List[Dict]
|
||
) -> str:
|
||
"""生成智能系统提示词"""
|
||
tools_desc = "\n".join([
|
||
f"- {t.get('name')}: {t.get('description', '')}"
|
||
for t in tools_config
|
||
])
|
||
|
||
return f'''你是 {agent_name}，一个专业的 AI 智能助手。
|
||
|
||
## 角色定位
|
||
{description}
|
||
|
||
## 可用工具
|
||
{tools_desc}
|
||
|
||
## 工作原则
|
||
1. 理解用户意图：仔细分析用户的请求，确保准确理解需求
|
||
2. 选择合适工具：根据需求选择最合适的工具来完成任务
|
||
3. 清晰反馈：以用户友好的方式呈现结果
|
||
4. 错误处理：遇到问题时提供有用的错误信息和建议
|
||
|
||
## 响应格式
|
||
- 对于数据查询：返回结构化的 JSON 数据
|
||
- 对于操作请求：返回操作状态和结果
|
||
- 始终使用中文与用户交流（除非用户使用其他语言）'''
|
||
|
||
def generate_api_server(self, agent_name: str, description: str, tools_config: List[Dict] = None) -> str:
|
||
"""
|
||
生成 API 服务器代码 (api_server.py)
|
||
基于 _template/src/server/api_server.py 框架
|
||
集成回调功能用于计费
|
||
|
||
Args:
|
||
agent_name: Agent 名称
|
||
description: Agent 描述
|
||
tools_config: 工具配置列表（用于生成工具名称列表）
|
||
"""
|
||
# 生成工具名称列表（用于回调）
|
||
tool_names = []
|
||
if tools_config:
|
||
for tool in tools_config:
|
||
tool_names.append(tool.get("name", "unknown_tool"))
|
||
|
||
tool_names_str = json.dumps(tool_names, ensure_ascii=False)
|
||
|
||
return f'''"""
|
||
HTTP API 服务器 - {agent_name}
|
||
|
||
{description}
|
||
提供 REST API 和 MCP HTTP/SSE 端点。
|
||
集成回调功能用于计费。
|
||
自动生成时间: {datetime.now().isoformat()}
|
||
"""
|
||
import json
|
||
import uuid
|
||
import os
|
||
import logging
|
||
from typing import Optional, Dict, Any, AsyncGenerator, List
|
||
from contextlib import asynccontextmanager
|
||
from datetime import datetime
|
||
|
||
from fastapi import FastAPI, HTTPException, Request, Header, Depends
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
from fastapi.responses import StreamingResponse, JSONResponse
|
||
from pydantic import BaseModel, Field
|
||
|
||
from .mcp_server import TOOL_MAP, TOOL_LIST
|
||
from .agent_callback_utils import AgentCallbackHandler, CallbackContextManager
|
||
|
||
# ==================== 配置 ====================
|
||
|
||
SERVER_NAME = "{agent_name}"
|
||
POD_NAME = os.getenv("POD_NAME", "{agent_name}")
|
||
USER_ID = os.getenv("USER_ID", "")
|
||
|
||
# 配置日志
|
||
logging.basicConfig(level=logging.INFO)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 本 Agent 包含的工具列表
|
||
AGENT_TOOLS = {tool_names_str}
|
||
|
||
# ==================== 回调处理器 ====================
|
||
|
||
callback_handler: Optional[AgentCallbackHandler] = None
|
||
|
||
|
||
def get_callback_handler() -> AgentCallbackHandler:
|
||
"""获取或创建回调处理器（单例）"""
|
||
global callback_handler
|
||
if callback_handler is None:
|
||
callback_handler = AgentCallbackHandler(
|
||
agent_name=POD_NAME,
|
||
user_id=USER_ID
|
||
)
|
||
return callback_handler
|
||
|
||
|
||
# ==================== FastAPI 应用 ====================
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(app: FastAPI):
|
||
logger.info(f"🚀 {{SERVER_NAME}} 启动")
|
||
logger.info(f" 包含工具: {{', '.join(AGENT_TOOLS)}}")
|
||
logger.info(f" Pod 名称: {{POD_NAME}}")
|
||
logger.info(f" 用户 ID: {{USER_ID or '未设置'}}")
|
||
yield
|
||
logger.info(f"🛑 {{SERVER_NAME}} 关闭")
|
||
|
||
app = FastAPI(
|
||
title=SERVER_NAME,
|
||
description="{description}",
|
||
version="1.0.0",
|
||
lifespan=lifespan
|
||
)
|
||
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=["*"],
|
||
allow_credentials=True,
|
||
allow_methods=["*"],
|
||
allow_headers=["*"],
|
||
)
|
||
|
||
|
||
# ==================== API Key 验证 ====================
|
||
|
||
async def verify_api_key(
|
||
api_key: Optional[str] = Header(None, alias="api-key"),
|
||
authorization: Optional[str] = Header(None)
|
||
) -> str:
|
||
"""验证 API Key"""
|
||
if api_key and api_key.strip() and api_key.strip() != "sk":
|
||
return api_key.strip()
|
||
|
||
if authorization:
|
||
key = authorization[7:].strip() if authorization.startswith("Bearer ") else authorization.strip()
|
||
if key and key != "sk":
|
||
return key
|
||
|
||
# 允许无 API Key 访问（使用默认配置）
|
||
return os.getenv("OPENAI_API_KEY", "sk")
|
||
|
||
|
||
def get_api_key_from_request(request: Request) -> Optional[str]:
|
||
"""从请求头提取 API Key（不验证）"""
|
||
api_key = request.headers.get("api-key") or request.headers.get("api_key")
|
||
if not api_key:
|
||
auth = request.headers.get("Authorization")
|
||
if auth:
|
||
api_key = auth[7:] if auth.startswith("Bearer ") else auth
|
||
return api_key or os.getenv("OPENAI_API_KEY", "sk")
|
||
|
||
|
||
# ==================== 健康检查 ====================
|
||
|
||
@app.get("/")
|
||
async def root():
|
||
return {{
|
||
"service": SERVER_NAME,
|
||
"status": "running",
|
||
"tools": list(TOOL_MAP.keys()),
|
||
"tools_count": len(TOOL_MAP),
|
||
"pod_name": POD_NAME
|
||
}}
|
||
|
||
|
||
@app.get("/health")
|
||
async def health():
|
||
return {{"status": "healthy", "service": SERVER_NAME, "tools_count": len(TOOL_MAP)}}
|
||
|
||
|
||
# ==================== MCP 端点 ====================
|
||
|
||
sessions: Dict[str, Dict] = {{}}
|
||
|
||
|
||
async def handle_mcp_request(data: Dict, session_id: str = None, api_key: str = None, user_id: str = None) -> Dict:
|
||
"""处理 MCP JSON-RPC 请求（带回调）"""
|
||
method = data.get("method")
|
||
params = data.get("params", {{}})
|
||
req_id = data.get("id")
|
||
|
||
try:
|
||
if method == "initialize":
|
||
session_id = session_id or str(uuid.uuid4())
|
||
sessions[session_id] = {{"initialized": True}}
|
||
return {{
|
||
"jsonrpc": "2.0", "id": req_id,
|
||
"result": {{
|
||
"protocolVersion": "2024-11-05",
|
||
"capabilities": {{"tools": {{}}}},
|
||
"serverInfo": {{"name": SERVER_NAME, "version": "1.0.0"}}
|
||
}}
|
||
}}
|
||
|
||
elif method == "tools/list":
|
||
return {{"jsonrpc": "2.0", "id": req_id, "result": {{"tools": TOOL_LIST}}}}
|
||
|
||
elif method == "tools/call":
|
||
tool_name = params.get("name")
|
||
args = params.get("arguments", {{}})
|
||
|
||
if tool_name not in TOOL_MAP:
|
||
raise ValueError(f"Unknown tool: {{tool_name}}")
|
||
|
||
# 设置 API Key 到环境变量
|
||
old_key = os.environ.get('OPENAI_API_KEY')
|
||
if api_key:
|
||
os.environ['OPENAI_API_KEY'] = api_key
|
||
|
||
# 使用回调上下文管理器（如果有 user_id）
|
||
effective_user_id = user_id or USER_ID
|
||
|
||
try:
|
||
if effective_user_id:
|
||
handler = get_callback_handler()
|
||
with CallbackContextManager(
|
||
handler=handler,
|
||
user_id=effective_user_id,
|
||
request_id=f"mcp-{{req_id}}-{{int(datetime.utcnow().timestamp())}}"
|
||
) as ctx:
|
||
ctx.add_tool(tool_name)
|
||
result = await TOOL_MAP[tool_name](**args)
|
||
else:
|
||
result = await TOOL_MAP[tool_name](**args)
|
||
finally:
|
||
if old_key:
|
||
os.environ['OPENAI_API_KEY'] = old_key
|
||
|
||
return {{
|
||
"jsonrpc": "2.0", "id": req_id,
|
||
"result": {{"content": [{{"type": "text", "text": str(result)}}]}}
|
||
}}
|
||
|
||
elif method == "ping":
|
||
return {{"jsonrpc": "2.0", "id": req_id, "result": {{}}}}
|
||
|
||
else:
|
||
raise ValueError(f"Unknown method: {{method}}")
|
||
|
||
except Exception as e:
|
||
return {{"jsonrpc": "2.0", "id": req_id, "error": {{"code": -32603, "message": str(e)}}}}
|
||
|
||
|
||
@app.post("/mcp")
|
||
async def mcp_endpoint(request: Request):
|
||
"""MCP HTTP 端点"""
|
||
try:
|
||
body = await request.json()
|
||
session_id = request.headers.get("x-mcp-session-id")
|
||
api_key = get_api_key_from_request(request)
|
||
user_id = request.headers.get("x-user-id") or USER_ID
|
||
response = await handle_mcp_request(body, session_id, api_key, user_id)
|
||
return JSONResponse(content=response, headers={{"x-mcp-session-id": session_id or ""}})
|
||
except Exception as e:
|
||
return JSONResponse(status_code=400, content={{"jsonrpc": "2.0", "error": {{"code": -32700, "message": str(e)}}}})
|
||
|
||
|
||
@app.get("/mcp/sse")
|
||
async def mcp_sse(request: Request):
|
||
"""MCP SSE 端点"""
|
||
session_id = request.headers.get("x-mcp-session-id") or str(uuid.uuid4())
|
||
|
||
async def stream() -> AsyncGenerator[str, None]:
|
||
yield f"data: {{json.dumps({{'type': 'connection', 'sessionId': session_id}})}}\\n\\n"
|
||
import asyncio
|
||
while True:
|
||
await asyncio.sleep(30)
|
||
yield f"data: {{json.dumps({{'type': 'ping'}})}}\\n\\n"
|
||
|
||
return StreamingResponse(stream(), media_type="text/event-stream",
|
||
headers={{"Cache-Control": "no-cache", "x-mcp-session-id": session_id}})
|
||
|
||
|
||
@app.post("/mcp/sse")
|
||
async def mcp_sse_post(request: Request):
|
||
"""MCP SSE POST 端点"""
|
||
try:
|
||
body = await request.json()
|
||
session_id = request.headers.get("x-mcp-session-id") or str(uuid.uuid4())
|
||
api_key = get_api_key_from_request(request)
|
||
user_id = request.headers.get("x-user-id") or USER_ID
|
||
|
||
async def stream() -> AsyncGenerator[str, None]:
|
||
response = await handle_mcp_request(body, session_id, api_key, user_id)
|
||
yield f"data: {{json.dumps(response)}}\\n\\n"
|
||
|
||
return StreamingResponse(stream(), media_type="text/event-stream",
|
||
headers={{"Cache-Control": "no-cache", "x-mcp-session-id": session_id}})
|
||
except Exception as e:
|
||
return JSONResponse(status_code=400, content={{"jsonrpc": "2.0", "error": {{"code": -32700, "message": str(e)}}}})
|
||
|
||
|
||
# ==================== 业务 API ====================
|
||
|
||
class ToolCallRequest(BaseModel):
|
||
"""工具调用请求"""
|
||
tool_name: str = Field(..., description="工具名称")
|
||
parameters: Dict[str, Any] = Field(default={{}}, description="工具参数")
|
||
user_id: Optional[str] = Field(None, description="用户ID（用于计费回调）")
|
||
|
||
|
||
class MultiToolCallRequest(BaseModel):
|
||
"""批量工具调用请求"""
|
||
calls: List[ToolCallRequest] = Field(..., description="工具调用列表")
|
||
user_id: Optional[str] = Field(None, description="用户ID（用于计费回调）")
|
||
|
||
|
||
class ToolCallResponse(BaseModel):
|
||
"""工具调用响应"""
|
||
success: bool
|
||
result: Optional[Any] = None
|
||
error: Optional[str] = None
|
||
tools_used: Optional[List[str]] = None
|
||
|
||
|
||
class MultiToolCallResponse(BaseModel):
|
||
"""批量工具调用响应"""
|
||
success: bool
|
||
results: List[ToolCallResponse]
|
||
tools_used: List[str]
|
||
total_calls: int
|
||
|
||
|
||
@app.get("/tools")
|
||
async def list_tools():
|
||
"""列出可用工具"""
|
||
return {{
|
||
"tools": [
|
||
{{"name": t["name"], "description": t["description"]}}
|
||
for t in TOOL_LIST
|
||
],
|
||
"count": len(TOOL_LIST)
|
||
}}
|
||
|
||
|
||
@app.post("/tools/call", response_model=ToolCallResponse)
|
||
async def call_tool(request: ToolCallRequest, api_key: str = Depends(verify_api_key)):
|
||
"""调用单个工具（带计费回调）"""
|
||
if request.tool_name not in TOOL_MAP:
|
||
raise HTTPException(status_code=404, detail=f"工具 {{request.tool_name}} 不存在")
|
||
|
||
effective_user_id = request.user_id or USER_ID
|
||
tools_used = [request.tool_name]
|
||
|
||
try:
|
||
old_key = os.environ.get('OPENAI_API_KEY')
|
||
os.environ['OPENAI_API_KEY'] = api_key
|
||
|
||
try:
|
||
# 使用回调上下文管理器
|
||
if effective_user_id:
|
||
handler = get_callback_handler()
|
||
with CallbackContextManager(
|
||
handler=handler,
|
||
user_id=effective_user_id,
|
||
request_id=f"api-{{int(datetime.utcnow().timestamp())}}"
|
||
) as ctx:
|
||
ctx.add_tool(request.tool_name)
|
||
result = await TOOL_MAP[request.tool_name](**request.parameters)
|
||
else:
|
||
result = await TOOL_MAP[request.tool_name](**request.parameters)
|
||
|
||
return ToolCallResponse(
|
||
success=True,
|
||
result=json.loads(result) if isinstance(result, str) else result,
|
||
tools_used=tools_used
|
||
)
|
||
finally:
|
||
if old_key:
|
||
os.environ['OPENAI_API_KEY'] = old_key
|
||
except Exception as e:
|
||
logger.error(f"工具调用失败: {{e}}")
|
||
return ToolCallResponse(success=False, error=str(e), tools_used=tools_used)
|
||
|
||
|
||
@app.post("/tools/batch-call", response_model=MultiToolCallResponse)
|
||
async def batch_call_tools(request: MultiToolCallRequest, api_key: str = Depends(verify_api_key)):
|
||
"""批量调用多个工具（带计费回调）"""
|
||
effective_user_id = request.user_id or USER_ID
|
||
results = []
|
||
tools_used = []
|
||
|
||
# 设置 API Key
|
||
old_key = os.environ.get('OPENAI_API_KEY')
|
||
os.environ['OPENAI_API_KEY'] = api_key
|
||
|
||
try:
|
||
# 使用回调上下文管理器
|
||
if effective_user_id:
|
||
handler = get_callback_handler()
|
||
with CallbackContextManager(
|
||
handler=handler,
|
||
user_id=effective_user_id,
|
||
request_id=f"batch-{{int(datetime.utcnow().timestamp())}}"
|
||
) as ctx:
|
||
for call in request.calls:
|
||
if call.tool_name not in TOOL_MAP:
|
||
results.append(ToolCallResponse(
|
||
success=False,
|
||
error=f"工具 {{call.tool_name}} 不存在"
|
||
))
|
||
continue
|
||
|
||
try:
|
||
ctx.add_tool(call.tool_name)
|
||
tools_used.append(call.tool_name)
|
||
result = await TOOL_MAP[call.tool_name](**call.parameters)
|
||
results.append(ToolCallResponse(
|
||
success=True,
|
||
result=json.loads(result) if isinstance(result, str) else result
|
||
))
|
||
except Exception as e:
|
||
results.append(ToolCallResponse(success=False, error=str(e)))
|
||
else:
|
||
for call in request.calls:
|
||
if call.tool_name not in TOOL_MAP:
|
||
results.append(ToolCallResponse(
|
||
success=False,
|
||
error=f"工具 {{call.tool_name}} 不存在"
|
||
))
|
||
continue
|
||
|
||
try:
|
||
tools_used.append(call.tool_name)
|
||
result = await TOOL_MAP[call.tool_name](**call.parameters)
|
||
results.append(ToolCallResponse(
|
||
success=True,
|
||
result=json.loads(result) if isinstance(result, str) else result
|
||
))
|
||
except Exception as e:
|
||
results.append(ToolCallResponse(success=False, error=str(e)))
|
||
finally:
|
||
if old_key:
|
||
os.environ['OPENAI_API_KEY'] = old_key
|
||
|
||
return MultiToolCallResponse(
|
||
success=all(r.success for r in results),
|
||
results=results,
|
||
tools_used=list(set(tools_used)),
|
||
total_calls=len(request.calls)
|
||
)
|
||
|
||
|
||
# ==================== 智能对话（Agent Chat）====================
|
||
|
||
# LLM 配置
|
||
LLM_BASE_URL = os.getenv("OPENAI_BASE_URL", "https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io/v1")
|
||
LLM_MODEL = os.getenv("MODEL_NAME", "taiji/gpt-4o-mini")
|
||
|
||
class ChatRequest(BaseModel):
|
||
"""聊天请求"""
|
||
message: str
|
||
conversation_id: Optional[str] = None
|
||
user_id: Optional[str] = None
|
||
stream: bool = False
|
||
|
||
class ChatResponse(BaseModel):
|
||
"""聊天响应"""
|
||
success: bool
|
||
message: str
|
||
tools_used: List[str] = []
|
||
conversation_id: Optional[str] = None
|
||
error: Optional[str] = None
|
||
|
||
# 对话历史存储
|
||
conversations: Dict[str, List[Dict]] = {{}}
|
||
|
||
def get_tools_description() -> str:
|
||
\"\"\"生成工具描述供 LLM 使用\"\"\"
|
||
tools_desc = []
|
||
for t in TOOL_LIST:
|
||
params = t.get("parameters", {{}})
|
||
param_desc = ", ".join([f"{{k}}: {{v.get('type', 'string')}}" for k, v in params.items()])
|
||
tools_desc.append(f"- {{t['name']}}: {{t['description']}}\\n 参数: {{param_desc or '无'}}")
|
||
return "\\n".join(tools_desc)
|
||
|
||
def build_system_prompt() -> str:
|
||
\"\"\"构建系统提示\"\"\"
|
||
tools_desc = get_tools_description()
|
||
json_example = '{{"action": "tool_call", "tool": "工具名称", "parameters": {{"参数名": "参数值"}}}}'
|
||
return f\"\"\"你是一个智能助手 {{SERVER_NAME}}，可以使用以下工具来帮助用户：
|
||
|
||
{{tools_desc}}
|
||
|
||
当用户的问题需要使用工具时，请按以下 JSON 格式回复：
|
||
{{json_example}}
|
||
|
||
当不需要工具时，直接回复用户的问题。
|
||
|
||
重要规则：
|
||
1. 如果问题可以用工具解决，优先使用工具
|
||
2. 工具调用必须严格使用上述 JSON 格式
|
||
3. 参数名必须与工具定义匹配
|
||
4. 一次只调用一个工具\"\"\"
|
||
|
||
async def call_llm(messages: List[Dict], api_key: str) -> str:
|
||
\"\"\"调用 LLM - 使用请求传入的 API Key（用于计费）\"\"\"
|
||
import httpx
|
||
|
||
if not api_key or api_key in ("sk", "sk-test", "test"):
|
||
raise ValueError("请提供有效的 API Key（用于计费）")
|
||
|
||
async with httpx.AsyncClient(timeout=60.0) as client:
|
||
response = await client.post(
|
||
f"{{LLM_BASE_URL}}/chat/completions",
|
||
headers={{
|
||
"Authorization": f"Bearer {{api_key}}",
|
||
"Content-Type": "application/json"
|
||
}},
|
||
json={{
|
||
"model": LLM_MODEL,
|
||
"messages": messages,
|
||
"temperature": 0.3,
|
||
"max_tokens": 2000
|
||
}}
|
||
)
|
||
response.raise_for_status()
|
||
return response.json()["choices"][0]["message"]["content"]
|
||
|
||
def parse_tool_call(response: str) -> Optional[Dict]:
|
||
\"\"\"解析 LLM 响应中的工具调用\"\"\"
|
||
# 方法1：尝试直接解析整个响应
|
||
try:
|
||
data = json.loads(response.strip())
|
||
if isinstance(data, dict) and data.get("action") == "tool_call":
|
||
return data
|
||
except json.JSONDecodeError:
|
||
pass
|
||
|
||
# 方法2：提取 JSON 块（处理 markdown 代码块）
|
||
import re
|
||
# 匹配 ```json ... ``` 或 ``` ... ```
|
||
code_block = re.search(r'```(?:json)?\\s*([\\s\\S]*?)```', response)
|
||
if code_block:
|
||
try:
|
||
data = json.loads(code_block.group(1).strip())
|
||
if isinstance(data, dict) and data.get("action") == "tool_call":
|
||
return data
|
||
except json.JSONDecodeError:
|
||
pass
|
||
|
||
# 方法3：查找 JSON 对象（从 {{ 到匹配的 }}）
|
||
start = response.find('{{')
|
||
if start == -1:
|
||
start = response.find('{{"{{"') # 处理转义
|
||
if start == -1:
|
||
start = response.find('{{"action"')
|
||
|
||
if start != -1:
|
||
# 找到平衡的 }}
|
||
depth = 0
|
||
end = start
|
||
for i, c in enumerate(response[start:]):
|
||
if c == '{{':
|
||
depth += 1
|
||
elif c == '}}':
|
||
depth -= 1
|
||
if depth == 0:
|
||
end = start + i + 1
|
||
break
|
||
|
||
try:
|
||
data = json.loads(response[start:end])
|
||
if isinstance(data, dict) and data.get("action") == "tool_call":
|
||
return data
|
||
except json.JSONDecodeError:
|
||
pass
|
||
|
||
return None
|
||
|
||
@app.post("/chat", response_model=ChatResponse)
|
||
async def chat(request: ChatRequest, api_key: str = Depends(verify_api_key)):
|
||
\"\"\"
|
||
智能对话端点 - Agent 自动选择并调用工具
|
||
|
||
输入自然语言，Agent 会：
|
||
1. 理解用户意图
|
||
2. 自动选择合适的工具
|
||
3. 执行工具并返回结果
|
||
\"\"\"
|
||
effective_user_id = request.user_id or USER_ID
|
||
tools_used = []
|
||
|
||
# 获取或创建对话历史
|
||
conv_id = request.conversation_id or str(uuid.uuid4())
|
||
if conv_id not in conversations:
|
||
conversations[conv_id] = []
|
||
|
||
# 构建消息
|
||
messages = [
|
||
{{"role": "system", "content": build_system_prompt()}}
|
||
]
|
||
messages.extend(conversations[conv_id])
|
||
messages.append({{"role": "user", "content": request.message}})
|
||
|
||
try:
|
||
# 调用 LLM
|
||
llm_response = await call_llm(messages, api_key)
|
||
|
||
# 检查是否需要调用工具
|
||
tool_call = parse_tool_call(llm_response)
|
||
|
||
if tool_call and tool_call.get("tool") in TOOL_MAP:
|
||
tool_name = tool_call["tool"]
|
||
tool_params = tool_call.get("parameters", {{}})
|
||
tools_used.append(tool_name)
|
||
|
||
logger.info(f"🔧 调用工具: {{tool_name}}, 参数: {{tool_params}}")
|
||
|
||
# 执行工具调用
|
||
if effective_user_id:
|
||
handler = get_callback_handler()
|
||
with CallbackContextManager(
|
||
handler=handler,
|
||
user_id=effective_user_id,
|
||
request_id=f"chat-{{int(datetime.utcnow().timestamp())}}"
|
||
) as ctx:
|
||
ctx.add_tool(tool_name)
|
||
tool_result = await TOOL_MAP[tool_name](**tool_params)
|
||
else:
|
||
tool_result = await TOOL_MAP[tool_name](**tool_params)
|
||
|
||
# 将工具结果发送给 LLM 生成最终回复
|
||
messages.append({{"role": "assistant", "content": llm_response}})
|
||
messages.append({{"role": "user", "content": f"工具 {{tool_name}} 返回结果：{{tool_result}}\\n\\n请根据这个结果回答用户的问题。"}})
|
||
|
||
final_response = await call_llm(messages, api_key)
|
||
|
||
# 保存对话历史
|
||
conversations[conv_id].append({{"role": "user", "content": request.message}})
|
||
conversations[conv_id].append({{"role": "assistant", "content": final_response}})
|
||
|
||
return ChatResponse(
|
||
success=True,
|
||
message=final_response,
|
||
tools_used=tools_used,
|
||
conversation_id=conv_id
|
||
)
|
||
else:
|
||
# 不需要工具，直接返回 LLM 回复
|
||
conversations[conv_id].append({{"role": "user", "content": request.message}})
|
||
conversations[conv_id].append({{"role": "assistant", "content": llm_response}})
|
||
|
||
return ChatResponse(
|
||
success=True,
|
||
message=llm_response,
|
||
tools_used=[],
|
||
conversation_id=conv_id
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"聊天失败: {{e}}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return ChatResponse(
|
||
success=False,
|
||
message="",
|
||
error=str(e),
|
||
conversation_id=conv_id
|
||
)
|
||
|
||
|
||
@app.get("/chat/history/{{conversation_id}}")
|
||
async def get_chat_history(conversation_id: str):
|
||
\"\"\"获取对话历史\"\"\"
|
||
if conversation_id not in conversations:
|
||
raise HTTPException(status_code=404, detail="对话不存在")
|
||
return {{"conversation_id": conversation_id, "messages": conversations[conversation_id]}}
|
||
|
||
|
||
@app.delete("/chat/history/{{conversation_id}}")
|
||
async def clear_chat_history(conversation_id: str):
|
||
\"\"\"清除对话历史\"\"\"
|
||
if conversation_id in conversations:
|
||
del conversations[conversation_id]
|
||
return {{"success": True, "message": "对话历史已清除"}}
|
||
|
||
|
||
if __name__ == '__main__':
|
||
import uvicorn
|
||
uvicorn.run(app, host="0.0.0.0", port=8000)
|
||
'''
|
||
|
||
def generate_run_api_server(self) -> str:
|
||
"""生成入口文件 run_api_server.py"""
|
||
return '''#!/usr/bin/env python
|
||
"""启动 API 服务器"""
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).parent))
|
||
|
||
if __name__ == '__main__':
|
||
from src.server.api_server import app
|
||
import uvicorn
|
||
import os
|
||
|
||
host = os.getenv('API_HOST', '0.0.0.0')
|
||
port = int(os.getenv('API_PORT', '8000'))
|
||
|
||
print(f"🚀 启动 Agent API: http://{host}:{port}")
|
||
uvicorn.run(app, host=host, port=port, log_level="info")
|
||
'''
|
||
|
||
def generate_dockerfile(self, agent_name: str) -> str:
|
||
"""生成 Dockerfile (简化版，避免 WAF 拦截)"""
|
||
# 使用简化的 Dockerfile，避免触发 WAF
|
||
return '''FROM python:3.11-slim
|
||
|
||
WORKDIR /app
|
||
|
||
COPY requirements.txt .
|
||
RUN pip install --no-cache-dir -r requirements.txt
|
||
|
||
COPY . .
|
||
|
||
ENV PORT=8000
|
||
EXPOSE 8000
|
||
|
||
CMD ["python", "run_api_server.py"]
|
||
'''
|
||
|
||
def generate_requirements(self) -> str:
|
||
"""生成 requirements.txt"""
|
||
return '''# Pydantic AI
|
||
pydantic-ai>=0.0.14
|
||
|
||
# MCP
|
||
mcp>=0.9.0
|
||
fastmcp>=0.1.0
|
||
|
||
# FastAPI
|
||
fastapi>=0.109.0
|
||
uvicorn[standard]>=0.27.0
|
||
|
||
# HTTP Client
|
||
httpx>=0.25.0
|
||
aiohttp>=3.9.0
|
||
requests>=2.31.0
|
||
|
||
# 其他
|
||
python-dotenv>=1.0.0
|
||
'''
|
||
|
||
def generate_callback_utils(self) -> str:
|
||
"""生成回调工具代码 (用于计费)"""
|
||
return '''"""
|
||
Agent回调工具 - 用于向Agent Manager回调运行时长记录（计费）
|
||
"""
|
||
import os
|
||
import time
|
||
import logging
|
||
import requests
|
||
from typing import Optional, List
|
||
from datetime import datetime, timezone
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class AgentCallbackHandler:
|
||
"""Agent回调处理器"""
|
||
|
||
def __init__(
|
||
self,
|
||
agent_name: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
callback_url: Optional[str] = None
|
||
):
|
||
"""
|
||
初始化回调处理器
|
||
|
||
Args:
|
||
agent_name: Agent名称，默认从环境变量 POD_NAME 获取
|
||
user_id: 用户ID，默认从环境变量 USER_ID 获取
|
||
callback_url: 回调URL，默认从环境变量 AGENT_CALLBACK_URL 获取
|
||
"""
|
||
self.agent_name = agent_name or os.getenv("POD_NAME", "unknown-agent")
|
||
self.user_id = user_id or os.getenv("USER_ID", "")
|
||
self.callback_url = callback_url or os.getenv(
|
||
"AGENT_CALLBACK_URL",
|
||
"http://mcp-server.taiji-ai.svc.cluster.local:8000/api/v1/billing/agent-callback"
|
||
)
|
||
|
||
self.start_time: Optional[datetime] = None
|
||
self.tools_used: List[str] = []
|
||
self.request_id: Optional[str] = None
|
||
|
||
logger.info(f"AgentCallbackHandler initialized: agent={self.agent_name}, callback_url={self.callback_url}")
|
||
|
||
def start_request(self, request_id: Optional[str] = None, user_id: Optional[str] = None):
|
||
"""
|
||
开始一次请求处理
|
||
|
||
Args:
|
||
request_id: 请求ID
|
||
user_id: 用户ID（如果提供则覆盖默认值）
|
||
"""
|
||
self.start_time = datetime.now(timezone.utc)
|
||
self.tools_used = []
|
||
self.request_id = request_id or f"req-{int(time.time())}"
|
||
|
||
if user_id:
|
||
self.user_id = user_id
|
||
|
||
logger.info(f"Request started: request_id={self.request_id}, user_id={self.user_id}")
|
||
|
||
def add_tool_used(self, tool_name: str):
|
||
"""
|
||
记录使用的工具
|
||
|
||
Args:
|
||
tool_name: 工具名称
|
||
"""
|
||
if tool_name not in self.tools_used:
|
||
self.tools_used.append(tool_name)
|
||
logger.debug(f"Tool used: {tool_name}")
|
||
|
||
def end_request(self, tools_used: Optional[List[str]] = None) -> bool:
|
||
"""
|
||
结束请求并发送回调
|
||
|
||
Args:
|
||
tools_used: 使用的工具列表（可选，如果提供则覆盖内部记录）
|
||
|
||
Returns:
|
||
是否成功发送回调
|
||
"""
|
||
if not self.start_time:
|
||
logger.warning("Cannot end request: no start time recorded")
|
||
return False
|
||
|
||
if not self.user_id:
|
||
logger.warning("Cannot send callback: user_id not set")
|
||
return False
|
||
|
||
end_time = datetime.now(timezone.utc)
|
||
running_time = (end_time - self.start_time).total_seconds()
|
||
|
||
# 使用提供的工具列表或内部记录
|
||
final_tools_used = tools_used if tools_used is not None else self.tools_used
|
||
|
||
# 发送回调
|
||
success = self._send_callback(
|
||
running_time_seconds=int(running_time),
|
||
start_time=self.start_time,
|
||
end_time=end_time,
|
||
tools_used=final_tools_used
|
||
)
|
||
|
||
# 重置状态
|
||
self.start_time = None
|
||
self.tools_used = []
|
||
self.request_id = None
|
||
|
||
return success
|
||
|
||
def _send_callback(
|
||
self,
|
||
running_time_seconds: int,
|
||
start_time: datetime,
|
||
end_time: datetime,
|
||
tools_used: List[str]
|
||
) -> bool:
|
||
"""
|
||
发送回调到Agent Manager
|
||
|
||
Args:
|
||
running_time_seconds: 运行时长（秒）
|
||
start_time: 开始时间
|
||
end_time: 结束时间
|
||
tools_used: 使用的工具列表
|
||
|
||
Returns:
|
||
是否成功发送
|
||
"""
|
||
try:
|
||
payload = {
|
||
"agentName": self.agent_name,
|
||
"userId": self.user_id,
|
||
"podRunningTimeSeconds": running_time_seconds,
|
||
"toolsUsed": tools_used,
|
||
"startTime": start_time.isoformat(),
|
||
"endTime": end_time.isoformat(),
|
||
"requestId": self.request_id
|
||
}
|
||
|
||
logger.info(f"Sending callback: {payload}")
|
||
|
||
response = requests.post(
|
||
self.callback_url,
|
||
json=payload,
|
||
timeout=5
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
logger.info(f"Callback sent successfully: {response.json()}")
|
||
return True
|
||
else:
|
||
logger.error(f"Callback failed with status {response.status_code}: {response.text}")
|
||
return False
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
logger.error(f"Failed to send callback: {str(e)}")
|
||
return False
|
||
except Exception as e:
|
||
logger.error(f"Unexpected error sending callback: {str(e)}")
|
||
return False
|
||
|
||
|
||
class CallbackContextManager:
|
||
"""回调上下文管理器 - 使用with语句自动处理开始和结束"""
|
||
|
||
def __init__(
|
||
self,
|
||
handler: AgentCallbackHandler,
|
||
request_id: Optional[str] = None,
|
||
user_id: Optional[str] = None,
|
||
tools_used: Optional[List[str]] = None
|
||
):
|
||
"""
|
||
初始化上下文管理器
|
||
|
||
Args:
|
||
handler: AgentCallbackHandler实例
|
||
request_id: 请求ID
|
||
user_id: 用户ID
|
||
tools_used: 使用的工具列表（可选）
|
||
"""
|
||
self.handler = handler
|
||
self.request_id = request_id
|
||
self.user_id = user_id
|
||
self.tools_used = tools_used or []
|
||
|
||
def __enter__(self):
|
||
"""进入上下文时开始计时"""
|
||
self.handler.start_request(
|
||
request_id=self.request_id,
|
||
user_id=self.user_id
|
||
)
|
||
return self
|
||
|
||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||
"""退出上下文时发送回调"""
|
||
self.handler.end_request(tools_used=self.tools_used)
|
||
return False # 不抑制异常
|
||
|
||
def add_tool(self, tool_name: str):
|
||
"""添加使用的工具"""
|
||
self.handler.add_tool_used(tool_name)
|
||
if tool_name not in self.tools_used:
|
||
self.tools_used.append(tool_name)
|
||
'''
|
||
|
||
def generate_gitea_action(
|
||
self,
|
||
agent_name: str,
|
||
auto_deploy: bool = True,
|
||
cpu_request: str = "100m",
|
||
cpu_limit: str = "500m",
|
||
memory_request: str = "128Mi",
|
||
memory_limit: str = "512Mi",
|
||
replicas: int = 1,
|
||
tool_api_keys: List[str] = None
|
||
) -> str:
|
||
"""
|
||
生成 Gitea Actions CI/CD 配置
|
||
参考: http://gitee.ath.cx:3000/xiaohei/cicd-AKS
|
||
使用预配置的密钥
|
||
|
||
Args:
|
||
agent_name: Agent 名称
|
||
auto_deploy: 是否自动部署
|
||
cpu_request: CPU 请求 (如 100m, 500m)
|
||
cpu_limit: CPU 限制 (如 500m, 1000m)
|
||
memory_request: 内存请求 (如 128Mi, 256Mi)
|
||
memory_limit: 内存限制 (如 512Mi, 1Gi)
|
||
replicas: 副本数量
|
||
tool_api_keys: 工具 API 密钥列表（将注入到容器环境变量）
|
||
"""
|
||
k8s_name = agent_name.lower().replace("_", "-").replace(" ", "-")
|
||
image_repo = f"{self.acr_namespace}/{k8s_name}"
|
||
|
||
# 生成工具 API Key 环境变量配置
|
||
tool_api_key_env = ""
|
||
if tool_api_keys:
|
||
# 如果只有一个 key，使用 TOOL_API_KEY
|
||
if len(tool_api_keys) == 1:
|
||
tool_api_key_env = f''' - name: TOOL_API_KEY
|
||
value: "{tool_api_keys[0]}"'''
|
||
else:
|
||
# 多个 key 时，使用编号
|
||
env_lines = []
|
||
for i, key in enumerate(tool_api_keys):
|
||
env_lines.append(f''' - name: TOOL_API_KEY_{i}
|
||
value: "{key}"''')
|
||
# 第一个 key 也设置为默认的 TOOL_API_KEY
|
||
env_lines.insert(0, f''' - name: TOOL_API_KEY
|
||
value: "{tool_api_keys[0]}"''')
|
||
tool_api_key_env = "\n".join(env_lines)
|
||
|
||
deploy_step = ""
|
||
if auto_deploy:
|
||
deploy_step = f'''
|
||
- name: Login Azure
|
||
run: |
|
||
az login --service-principal \\
|
||
-u "$AZ_CLIENT_ID" \\
|
||
-p "$AZ_CLIENT_SECRET" \\
|
||
--tenant "$AZ_TENANT_ID"
|
||
az account set --subscription "$AZ_SUBSCRIPTION_ID"
|
||
|
||
- name: Get AKS Credentials
|
||
run: |
|
||
az aks get-credentials \\
|
||
--resource-group "$AZ_RG" \\
|
||
--name "$AZ_AKS" \\
|
||
--overwrite-existing
|
||
|
||
- name: Setup ACR Pull Secret
|
||
run: |
|
||
# 创建 Agent 专用命名空间
|
||
kubectl create namespace agent-{k8s_name} --dry-run=client -o yaml | kubectl apply -f -
|
||
|
||
# 创建或更新 ACR 拉取 Secret
|
||
kubectl create secret docker-registry acr-secret \\
|
||
--docker-server=$ACR_LOGIN_SERVER \\
|
||
--docker-username=$ACR_USERNAME \\
|
||
--docker-password=$ACR_PASSWORD \\
|
||
-n agent-{k8s_name} --dry-run=client -o yaml | kubectl apply -f -
|
||
|
||
- name: Deploy to AKS
|
||
run: |
|
||
# 删除旧的 deployment（如果存在）
|
||
kubectl delete deployment {k8s_name} -n agent-{k8s_name} --ignore-not-found=true
|
||
|
||
# 创建 deployment YAML
|
||
cat <<EOF | kubectl apply -f -
|
||
apiVersion: apps/v1
|
||
kind: Deployment
|
||
metadata:
|
||
name: {k8s_name}
|
||
namespace: agent-{k8s_name}
|
||
labels:
|
||
app: {k8s_name}
|
||
managed-by: gitea-cicd
|
||
spec:
|
||
replicas: {replicas}
|
||
selector:
|
||
matchLabels:
|
||
app: {k8s_name}
|
||
template:
|
||
metadata:
|
||
labels:
|
||
app: {k8s_name}
|
||
spec:
|
||
imagePullSecrets:
|
||
- name: acr-secret
|
||
nodeSelector:
|
||
kubernetes.io/arch: arm64
|
||
tolerations:
|
||
- key: "kubernetes.io/arch"
|
||
operator: "Equal"
|
||
value: "arm64"
|
||
effect: "NoSchedule"
|
||
containers:
|
||
- name: {k8s_name}
|
||
image: $ACR_LOGIN_SERVER/{image_repo}:${{{{ github.sha }}}}
|
||
ports:
|
||
- containerPort: 8000
|
||
env:
|
||
- name: OPENAI_BASE_URL
|
||
value: "https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io/v1"
|
||
- name: MODEL_NAME
|
||
value: "taiji/gpt-4o-mini"
|
||
{tool_api_key_env}
|
||
resources:
|
||
requests:
|
||
cpu: "{cpu_request}"
|
||
memory: "{memory_request}"
|
||
limits:
|
||
cpu: "{cpu_limit}"
|
||
memory: "{memory_limit}"
|
||
livenessProbe:
|
||
httpGet:
|
||
path: /health
|
||
port: 8000
|
||
initialDelaySeconds: 30
|
||
periodSeconds: 30
|
||
readinessProbe:
|
||
httpGet:
|
||
path: /health
|
||
port: 8000
|
||
initialDelaySeconds: 10
|
||
periodSeconds: 10
|
||
EOF
|
||
|
||
# 创建 Service
|
||
cat <<EOF | kubectl apply -f -
|
||
apiVersion: v1
|
||
kind: Service
|
||
metadata:
|
||
name: {k8s_name}-service
|
||
namespace: agent-{k8s_name}
|
||
spec:
|
||
type: LoadBalancer
|
||
ports:
|
||
- port: 80
|
||
targetPort: 8000
|
||
selector:
|
||
app: {k8s_name}
|
||
EOF
|
||
|
||
- name: Wait for Deployment
|
||
run: |
|
||
kubectl rollout status deployment/{k8s_name} -n agent-{k8s_name} --timeout=180s || {{
|
||
echo "=== Deployment 状态 ==="
|
||
kubectl get deployment {k8s_name} -n agent-{k8s_name} -o wide
|
||
echo "=== Pod 状态 ==="
|
||
kubectl get pods -l app={k8s_name} -n agent-{k8s_name} -o wide
|
||
echo "=== Pod 事件 ==="
|
||
kubectl describe pods -l app={k8s_name} -n agent-{k8s_name} | tail -30
|
||
exit 1
|
||
}}
|
||
|
||
echo "=== 部署成功 ==="
|
||
kubectl get svc {k8s_name}-service -n agent-{k8s_name}
|
||
|
||
# 等待 LoadBalancer IP
|
||
echo "等待 LoadBalancer IP..."
|
||
for i in {{1..30}}; do
|
||
EXTERNAL_IP=$(kubectl get svc {k8s_name}-service -n agent-{k8s_name} -o jsonpath='{{.status.loadBalancer.ingress[0].ip}}' 2>/dev/null)
|
||
if [ -n "$EXTERNAL_IP" ]; then
|
||
echo "✅ Agent LoadBalancer IP: $EXTERNAL_IP"
|
||
echo "EXTERNAL_IP=$EXTERNAL_IP" >> $GITHUB_ENV
|
||
break
|
||
fi
|
||
sleep 10
|
||
done
|
||
|
||
- name: Create DNS Record
|
||
if: env.EXTERNAL_IP != ''
|
||
run: |
|
||
echo "创建 DNS 记录: {k8s_name}.$AZURE_DNS_ZONE -> $EXTERNAL_IP"
|
||
|
||
# 获取 Azure 访问令牌
|
||
ACCESS_TOKEN=$(az account get-access-token --query accessToken -o tsv)
|
||
|
||
# 创建 DNS A 记录 (DNS Zone 在单独的资源组中)
|
||
DNS_RG="${{AZURE_DNS_RG:-taiji-Ai-v0}}"
|
||
DNS_URL="https://management.azure.com/subscriptions/$AZ_SUBSCRIPTION_ID/resourceGroups/$DNS_RG/providers/Microsoft.Network/dnsZones/$AZURE_DNS_ZONE/A/{k8s_name}?api-version=2018-05-01"
|
||
|
||
curl -X PUT "$DNS_URL" \\
|
||
-H "Authorization: Bearer $ACCESS_TOKEN" \\
|
||
-H "Content-Type: application/json" \\
|
||
-d '{{
|
||
"properties": {{
|
||
"TTL": 300,
|
||
"ARecords": [{{"ipv4Address": "'"$EXTERNAL_IP"'"}}]
|
||
}}
|
||
}}'
|
||
|
||
echo ""
|
||
echo "✅ DNS 记录创建成功!"
|
||
echo "🌐 Agent 访问地址: http://{k8s_name}.$AZURE_DNS_ZONE"
|
||
echo "🌐 Agent IP 地址: http://$EXTERNAL_IP"
|
||
'''
|
||
|
||
return f'''name: CI-CD-AKS-Agent
|
||
|
||
on:
|
||
push:
|
||
branches: ["main"]
|
||
workflow_dispatch:
|
||
|
||
jobs:
|
||
build_and_deploy:
|
||
runs-on: self-hosted
|
||
|
||
env:
|
||
ACR_LOGIN_SERVER: ${{{{ secrets.ACR_LOGIN_SERVER }}}}
|
||
ACR_USERNAME: ${{{{ secrets.ACR_USERNAME }}}}
|
||
ACR_PASSWORD: ${{{{ secrets.ACR_PASSWORD }}}}
|
||
AZ_CLIENT_ID: ${{{{ secrets.AZ_CLIENT_ID }}}}
|
||
AZ_CLIENT_SECRET: ${{{{ secrets.AZ_CLIENT_SECRET }}}}
|
||
AZ_TENANT_ID: ${{{{ secrets.AZ_TENANT_ID }}}}
|
||
AZ_SUBSCRIPTION_ID: ${{{{ secrets.AZ_SUBSCRIPTION_ID }}}}
|
||
AZ_RG: ${{{{ secrets.AZ_RG }}}}
|
||
AZ_AKS: ${{{{ secrets.AZ_AKS }}}}
|
||
AZURE_DNS_ZONE: ${{{{ secrets.AZURE_DNS_ZONE }}}}
|
||
AZURE_DNS_RG: ${{{{ secrets.AZURE_DNS_RG }}}}
|
||
|
||
steps:
|
||
- uses: actions/checkout@v4
|
||
|
||
- name: Install Azure CLI
|
||
run: |
|
||
if ! command -v az &> /dev/null; then
|
||
curl -sL https://aka.ms/InstallAzureCLIDeb | sudo bash
|
||
fi
|
||
|
||
- name: Install kubectl
|
||
run: |
|
||
if ! command -v kubectl &> /dev/null; then
|
||
ARCH=$(uname -m)
|
||
case $ARCH in
|
||
x86_64) ARCH="amd64" ;;
|
||
aarch64|arm64) ARCH="arm64" ;;
|
||
esac
|
||
curl -LO "https://dl.k8s.io/release/$(curl -L -s https://dl.k8s.io/release/stable.txt)/bin/linux/${{ARCH}}/kubectl"
|
||
sudo install -o root -g root -m 0755 kubectl /usr/local/bin/kubectl
|
||
rm kubectl
|
||
fi
|
||
|
||
- name: Build Docker image (ARM64)
|
||
run: |
|
||
docker build -t $ACR_LOGIN_SERVER/{image_repo}:${{{{ github.sha }}}} .
|
||
docker tag $ACR_LOGIN_SERVER/{image_repo}:${{{{ github.sha }}}} $ACR_LOGIN_SERVER/{image_repo}:latest
|
||
|
||
- name: Debug Secrets
|
||
run: |
|
||
echo "ACR_LOGIN_SERVER=$ACR_LOGIN_SERVER"
|
||
echo "ACR_USERNAME=$ACR_USERNAME"
|
||
echo "ACR_PASSWORD length=$(echo -n "$ACR_PASSWORD" | wc -c)"
|
||
if [ -z "$ACR_PASSWORD" ]; then echo "ERROR: ACR_PASSWORD is empty!"; fi
|
||
|
||
- name: Login to ACR
|
||
run: |
|
||
echo "$ACR_PASSWORD" | docker login "$ACR_LOGIN_SERVER" -u "$ACR_USERNAME" --password-stdin
|
||
|
||
- name: Push Docker image
|
||
run: |
|
||
docker push $ACR_LOGIN_SERVER/{image_repo}:${{{{ github.sha }}}}
|
||
docker push $ACR_LOGIN_SERVER/{image_repo}:latest
|
||
|
||
echo "✅ 镜像已推送: $ACR_LOGIN_SERVER/{image_repo}:latest"
|
||
{deploy_step}
|
||
'''
|
||
|
||
def generate_readme(
|
||
self,
|
||
agent_name: str,
|
||
description: str,
|
||
tools: List[Dict],
|
||
auto_deploy: bool = True
|
||
) -> str:
|
||
"""生成 README.md"""
|
||
tools_doc = "\n".join([f"- **{t.get('name')}**: {t.get('description', '')}" for t in tools])
|
||
k8s_name = agent_name.lower().replace("_", "-").replace(" ", "-")
|
||
|
||
deploy_doc = ""
|
||
if auto_deploy:
|
||
deploy_doc = f'''
|
||
## 自动部署
|
||
|
||
CI/CD 会自动：
|
||
1. 构建 Docker 镜像 (ARM64)
|
||
2. 推送到 Azure Container Registry
|
||
3. 部署到 AKS 集群
|
||
4. 创建 DNS 记录
|
||
|
||
部署后可通过以下方式访问：
|
||
- 命名空间: `agent-{k8s_name}`
|
||
- 服务: `{k8s_name}-service`
|
||
- **公网域名**: `http://{k8s_name}.taijiagnet.com`
|
||
'''
|
||
|
||
return f'''# {agent_name}
|
||
|
||
{description}
|
||
|
||
## 自动生成信息
|
||
|
||
- 生成时间: {datetime.now().isoformat()}
|
||
- 生成方式: Agent Manager 动态生成
|
||
- 框架: agent_templates/agents/_template
|
||
- 参考: [cicd-AKS](http://gitee.ath.cx:3000/xiaohei/cicd-AKS)
|
||
|
||
## 包含工具
|
||
|
||
{tools_doc}
|
||
|
||
## API 端点
|
||
|
||
### REST API
|
||
- `GET /` - 服务信息
|
||
- `GET /health` - 健康检查
|
||
- `GET /tools` - 列出可用工具
|
||
- `POST /tools/call` - 调用工具
|
||
|
||
### MCP 端点
|
||
- `POST /mcp` - MCP HTTP 端点
|
||
- `GET /mcp/sse` - MCP SSE 端点
|
||
- `POST /mcp/sse` - MCP SSE POST 端点
|
||
|
||
## 调用示例
|
||
|
||
```bash
|
||
# 健康检查
|
||
curl http://localhost:8000/health
|
||
|
||
# 列出工具
|
||
curl http://localhost:8000/tools
|
||
|
||
# 调用工具
|
||
curl -X POST http://localhost:8000/tools/call \\
|
||
-H "Content-Type: application/json" \\
|
||
-H "api-key: your-api-key" \\
|
||
-d '{{"tool_name": "example_tool", "parameters": {{}}}}'
|
||
|
||
# MCP 调用
|
||
curl -X POST http://localhost:8000/mcp \\
|
||
-H "Content-Type: application/json" \\
|
||
-d '{{"jsonrpc": "2.0", "method": "tools/list", "id": 1}}'
|
||
```
|
||
|
||
## 环境变量
|
||
|
||
- `API_PORT`: 服务端口 (默认: 8000)
|
||
- `API_HOST`: 服务地址 (默认: 0.0.0.0)
|
||
- `OPENAI_API_KEY`: AI 模型 API Key
|
||
- `OPENAI_BASE_URL`: AI 模型 API 端点
|
||
- `MODEL_NAME`: AI 模型名称
|
||
- `TOOL_API_KEY`: 工具 API 密钥 (如需要)
|
||
|
||
## CI/CD 配置
|
||
|
||
本仓库配置了 Gitea Actions (self-hosted runner)，推送到 main 分支后会自动：
|
||
1. 构建 Docker 镜像 (ARM64)
|
||
2. 推送到 Azure Container Registry
|
||
3. {"部署到 AKS 集群" if auto_deploy else "等待手动部署"}
|
||
|
||
### 必需的 Secrets
|
||
|
||
在 Gitee 仓库设置中配置以下 Secrets:
|
||
|
||
| Secret 名称 | 说明 | 值 |
|
||
|-------------|------|-----|
|
||
| `ACR_LOGIN_SERVER` | ACR 登录服务器 | `agnettaiji.azurecr.io` |
|
||
| `ACR_USERNAME` | ACR 用户名 | `agnettaiji` |
|
||
| `ACR_PASSWORD` | ACR 密码 | *(配置为仓库Secret)* |
|
||
| `AZ_CLIENT_ID` | Azure 服务主体 Client ID | `fb306798-2cfe-4ac9-ba48-eab7bc71bcfe` |
|
||
| `AZ_CLIENT_SECRET` | Azure 服务主体密钥 | *(配置为仓库Secret)* |
|
||
| `AZ_TENANT_ID` | Azure 租户 ID | `263c3ff6-1be5-4141-8308-b188464fb297` |
|
||
| `AZ_SUBSCRIPTION_ID` | Azure 订阅 ID | `c6c47e4c-f5f4-49f8-b26f-7728862c17d6` |
|
||
| `AZ_RG` | AKS 资源组名称 | `taiji-ai-pda` |
|
||
| `AZ_AKS` | AKS 集群名称 | `taiji-ai-pda` |
|
||
{deploy_doc}
|
||
|
||
## 许可证
|
||
|
||
MIT License
|
||
'''
|
||
|
||
def generate_full_project(
|
||
self,
|
||
agent_name: str,
|
||
description: str,
|
||
tools_config: List[Dict],
|
||
auto_deploy: bool = True,
|
||
cpu_request: str = "100m",
|
||
cpu_limit: str = "500m",
|
||
memory_request: str = "128Mi",
|
||
memory_limit: str = "512Mi",
|
||
replicas: int = 1
|
||
) -> Dict[str, str]:
|
||
"""
|
||
生成完整的 Agent 项目文件
|
||
基于 agent_templates/agents/_template 框架
|
||
包含回调功能用于计费
|
||
|
||
Args:
|
||
agent_name: Agent 名称
|
||
description: Agent 描述
|
||
tools_config: 工具配置列表（支持多个工具）
|
||
auto_deploy: 是否自动部署到 AKS
|
||
cpu_request: CPU 请求 (如 100m, 500m)
|
||
cpu_limit: CPU 限制 (如 500m, 1000m)
|
||
memory_request: 内存请求 (如 128Mi, 256Mi)
|
||
memory_limit: 内存限制 (如 512Mi, 1Gi)
|
||
replicas: 副本数量
|
||
|
||
Returns:
|
||
文件字典 {path: content}
|
||
"""
|
||
files = {}
|
||
|
||
k8s_name = agent_name.lower().replace("_", "-").replace(" ", "-")
|
||
|
||
# 生成 src/server/mcp_server.py
|
||
files["src/server/mcp_server.py"] = self.generate_mcp_server(
|
||
agent_name, description, tools_config
|
||
)
|
||
|
||
# 生成 src/server/api_server.py（传递 tools_config 用于回调）
|
||
files["src/server/api_server.py"] = self.generate_api_server(
|
||
agent_name, description, tools_config
|
||
)
|
||
|
||
# 生成 src/server/agent_callback_utils.py（回调工具）
|
||
files["src/server/agent_callback_utils.py"] = self.generate_callback_utils()
|
||
|
||
# 生成 src/server/__init__.py
|
||
files["src/server/__init__.py"] = '"""服务器模块"""\n'
|
||
|
||
# 生成 src/__init__.py
|
||
files["src/__init__.py"] = '"""源码模块"""\n'
|
||
|
||
# 生成入口文件
|
||
files["run_api_server.py"] = self.generate_run_api_server()
|
||
|
||
# 生成 Dockerfile
|
||
files["Dockerfile"] = self.generate_dockerfile(agent_name)
|
||
|
||
# 生成 requirements.txt
|
||
files["requirements.txt"] = self.generate_requirements()
|
||
|
||
# 从工具配置中提取 API Keys
|
||
tool_api_keys = []
|
||
for tool in tools_config:
|
||
auth = tool.get("auth")
|
||
if auth:
|
||
# 兼容 token 和 key 字段
|
||
api_key = auth.get("token") or auth.get("key")
|
||
if api_key:
|
||
tool_api_keys.append(api_key)
|
||
|
||
# 生成 CI/CD 配置
|
||
files[".gitea/workflows/ci-cd.yaml"] = self.generate_gitea_action(
|
||
agent_name=k8s_name,
|
||
auto_deploy=auto_deploy,
|
||
cpu_request=cpu_request,
|
||
cpu_limit=cpu_limit,
|
||
memory_request=memory_request,
|
||
memory_limit=memory_limit,
|
||
replicas=replicas,
|
||
tool_api_keys=tool_api_keys if tool_api_keys else None
|
||
)
|
||
|
||
# 生成 README
|
||
files["README.md"] = self.generate_readme(agent_name, description, tools_config, auto_deploy)
|
||
|
||
logger.info(f"生成 Agent 项目: {agent_name}")
|
||
logger.info(f" 包含工具: {len(tools_config)} 个")
|
||
logger.info(f" 工具列表: {[t.get('name') for t in tools_config]}")
|
||
logger.info(f" 自动部署: {auto_deploy}")
|
||
logger.info(f" 包含回调功能: 是")
|
||
|
||
return files
|
||
|
||
def get_cicd_secrets_info(self) -> Dict[str, str]:
|
||
"""获取 CI/CD 密钥信息（用于配置仓库 Secrets）"""
|
||
return self.cicd_secrets.copy()
|
||
|
||
|
||
# 全局实例
|
||
agent_code_generator = AgentCodeGenerator()
|