From 3ca3864e6d23cabdb6c640b0e6b7759ca28d2413 Mon Sep 17 00:00:00 2001 From: zhanggangyong Date: Thu, 15 Jan 2026 15:14:15 +0000 Subject: [PATCH] Add A2A AI Search Agent project --- search_agent/a2a-ai-search-agent/.gitignore | 24 + search_agent/a2a-ai-search-agent/Dockerfile | 24 + .../a2a-ai-search-agent/a2a_server.py | 487 ++++++++++++++++++ search_agent/a2a-ai-search-agent/agent.py | 333 ++++++++++++ search_agent/a2a-ai-search-agent/config.py | 141 +++++ search_agent/a2a-ai-search-agent/deploy.sh | 27 + .../a2a-ai-search-agent/k8s/configmap.yaml | 11 + .../a2a-ai-search-agent/k8s/deployment.yaml | 70 +++ .../a2a-ai-search-agent/k8s/secret.yaml | 13 + .../a2a-ai-search-agent/k8s/service.yaml | 18 + .../a2a-ai-search-agent/requirements.txt | 7 + 11 files changed, 1155 insertions(+) create mode 100644 search_agent/a2a-ai-search-agent/.gitignore create mode 100644 search_agent/a2a-ai-search-agent/Dockerfile create mode 100644 search_agent/a2a-ai-search-agent/a2a_server.py create mode 100644 search_agent/a2a-ai-search-agent/agent.py create mode 100644 search_agent/a2a-ai-search-agent/config.py create mode 100755 search_agent/a2a-ai-search-agent/deploy.sh create mode 100644 search_agent/a2a-ai-search-agent/k8s/configmap.yaml create mode 100644 search_agent/a2a-ai-search-agent/k8s/deployment.yaml create mode 100644 search_agent/a2a-ai-search-agent/k8s/secret.yaml create mode 100644 search_agent/a2a-ai-search-agent/k8s/service.yaml create mode 100644 search_agent/a2a-ai-search-agent/requirements.txt diff --git a/search_agent/a2a-ai-search-agent/.gitignore b/search_agent/a2a-ai-search-agent/.gitignore new file mode 100644 index 0000000..1a1a74a --- /dev/null +++ b/search_agent/a2a-ai-search-agent/.gitignore @@ -0,0 +1,24 @@ +# Python +__pycache__/ +*.py[cod] +*.class +*.so +.Python +venv/ +.venv/ +ENV/ +env/ + +# IDE +.vscode/ +.idea/ +*.swp +*.swo + +# 环境变量 +.env +*.log + +# 系统文件 +.DS_Store +Thumbs.db diff --git a/search_agent/a2a-ai-search-agent/Dockerfile b/search_agent/a2a-ai-search-agent/Dockerfile new file mode 100644 index 0000000..0753eeb --- /dev/null +++ b/search_agent/a2a-ai-search-agent/Dockerfile @@ -0,0 +1,24 @@ +FROM python:3.12-slim + +WORKDIR /app + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 + +RUN apt-get update && apt-get install -y --no-install-recommends curl \ + && rm -rf /var/lib/apt/lists/* + +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY config.py agent.py a2a_server.py ./ + +RUN useradd -m -u 1000 agent && chown -R agent:agent /app +USER agent + +EXPOSE 9000 + +HEALTHCHECK --interval=30s --timeout=10s --start-period=10s --retries=3 \ + CMD curl -f http://localhost:9000/health || exit 1 + +CMD ["uvicorn", "a2a_server:app", "--host", "0.0.0.0", "--port", "9000"] diff --git a/search_agent/a2a-ai-search-agent/a2a_server.py b/search_agent/a2a-ai-search-agent/a2a_server.py new file mode 100644 index 0000000..4e9d2c6 --- /dev/null +++ b/search_agent/a2a-ai-search-agent/a2a_server.py @@ -0,0 +1,487 @@ +""" +A2A协议兼容的Agent服务 +实现Google Agent2Agent协议规范 +""" + +import asyncio +import json +import uuid +from typing import Optional, Dict, Any, AsyncGenerator +from datetime import datetime +from contextlib import asynccontextmanager + +from fastapi import FastAPI, HTTPException, Request, Response +from fastapi.responses import StreamingResponse, JSONResponse +from fastapi.middleware.cors import CORSMiddleware +from pydantic import BaseModel, Field +import structlog + +from agent import LiteLLMAgent +from config import get_config, AgentConfig, A2AConfig + +# 配置日志 +logger = structlog.get_logger() + + +# ============== A2A 协议数据模型 ============== + +class A2APart(BaseModel): + """A2A消息部分""" + kind: str = "text" + text: Optional[str] = None + data: Optional[Dict[str, Any]] = None + mime_type: Optional[str] = None + + +class A2AMessage(BaseModel): + """A2A消息""" + role: str + parts: list[A2APart] + messageId: str = Field(default_factory=lambda: uuid.uuid4().hex) + + +class A2AMessageSendParams(BaseModel): + """A2A发送消息参数""" + message: A2AMessage + configuration: Optional[Dict[str, Any]] = None + + +class A2ARequest(BaseModel): + """A2A JSON-RPC请求""" + jsonrpc: str = "2.0" + id: str + method: str + params: Optional[Dict[str, Any]] = None + + +class A2AArtifact(BaseModel): + """A2A响应工件""" + artifactId: str = Field(default_factory=lambda: uuid.uuid4().hex) + name: str = "response" + parts: list[A2APart] + + +class A2ATaskStatus(BaseModel): + """A2A任务状态""" + state: str # submitted, working, input-required, completed, failed, canceled + timestamp: str = Field(default_factory=lambda: datetime.utcnow().isoformat() + "Z") + message: Optional[str] = None + + +class A2ATask(BaseModel): + """A2A任务""" + kind: str = "task" + id: str = Field(default_factory=lambda: uuid.uuid4().hex) + contextId: str = Field(default_factory=lambda: uuid.uuid4().hex) + status: A2ATaskStatus + artifacts: Optional[list[A2AArtifact]] = None + + +class A2AResponse(BaseModel): + """A2A JSON-RPC响应""" + jsonrpc: str = "2.0" + id: str + result: Optional[A2ATask] = None + error: Optional[Dict[str, Any]] = None + + +class A2AStreamEvent(BaseModel): + """A2A流式事件""" + kind: str + taskId: str + contextId: str + data: Optional[Dict[str, Any]] = None + + +# ============== Agent Card ============== + +class AgentSkill(BaseModel): + """Agent技能""" + id: str + name: str + description: str + inputSchema: Optional[Dict[str, Any]] = None + outputSchema: Optional[Dict[str, Any]] = None + + +class AgentCapabilities(BaseModel): + """Agent能力""" + text: bool = True + streaming: bool = True + push_notifications: bool = False + forms: bool = False + files: bool = False + + +class AgentCard(BaseModel): + """A2A Agent Card - 描述Agent能力""" + name: str + description: str + version: str + url: str + capabilities: AgentCapabilities + skills: list[AgentSkill] + authentication: Optional[Dict[str, Any]] = None + + +# ============== A2A Server ============== + +class A2AAgentServer: + """A2A协议Agent服务器""" + + def __init__( + self, + api_key: Optional[str] = None, + model: Optional[str] = None + ): + """ + 初始化A2A Agent服务器 + + Args: + api_key: LiteLLM API密钥 + model: 模型名称 + """ + # 获取配置 + self.llm_config, self.agent_config, self.a2a_config = get_config(api_key, model) + + # 创建Agent + self.agent = LiteLLMAgent( + litellm_config=self.llm_config, + agent_config=self.agent_config + ) + + # 任务存储 + self.tasks: Dict[str, A2ATask] = {} + + # 创建FastAPI应用 + self.app = self._create_app() + + def _create_app(self) -> FastAPI: + """创建FastAPI应用""" + + @asynccontextmanager + async def lifespan(app: FastAPI): + logger.info("A2A Agent服务启动", agent_name=self.agent_config.name) + yield + await self.agent.close() + logger.info("A2A Agent服务关闭") + + app = FastAPI( + title=f"{self.agent_config.name} - A2A Agent", + description=self.agent_config.description, + version=self.agent_config.version, + lifespan=lifespan + ) + + # CORS中间件 + app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], + ) + + # 注册路由 + self._register_routes(app) + + return app + + def _register_routes(self, app: FastAPI): + """注册A2A协议路由""" + + @app.get("/") + async def root(): + """服务根路径""" + return { + "name": self.agent_config.name, + "version": self.agent_config.version, + "protocol": "A2A", + "status": "running" + } + + @app.get("/.well-known/agent.json") + async def get_agent_card(request: Request): + """获取Agent Card (A2A发现协议)""" + base_url = str(request.base_url).rstrip("/") + + card = AgentCard( + name=self.agent_config.name, + description=self.agent_config.description, + version=self.agent_config.version, + url=base_url, + capabilities=AgentCapabilities( + text=True, + streaming=self.agent_config.enable_streaming, + push_notifications=False + ), + skills=[ + AgentSkill( + id="general-assistant", + name="通用助手", + description="回答问题、提供建议、协助完成各种任务" + ), + AgentSkill( + id="code-helper", + name="代码助手", + description="编写、解释和调试代码" + ) + ] + ) + return card.model_dump() + + @app.post("/message/send") + async def send_message(request: Request): + """A2A message/send 端点""" + body = await request.json() + + # 解析JSON-RPC请求 + try: + rpc_request = A2ARequest(**body) + except Exception as e: + return JSONResponse({ + "jsonrpc": "2.0", + "id": body.get("id", "unknown"), + "error": { + "code": -32600, + "message": f"Invalid Request: {str(e)}" + } + }) + + # 处理 message/send 方法 + if rpc_request.method == "message/send": + return await self._handle_message_send(rpc_request) + elif rpc_request.method == "message/stream": + return await self._handle_message_stream(rpc_request) + else: + return JSONResponse({ + "jsonrpc": "2.0", + "id": rpc_request.id, + "error": { + "code": -32601, + "message": f"Method not found: {rpc_request.method}" + } + }) + + @app.post("/message/stream") + async def stream_message(request: Request): + """A2A message/stream 端点 (SSE流式响应)""" + body = await request.json() + + try: + rpc_request = A2ARequest(**body) + except Exception as e: + return JSONResponse({ + "jsonrpc": "2.0", + "id": body.get("id", "unknown"), + "error": { + "code": -32600, + "message": f"Invalid Request: {str(e)}" + } + }) + + return await self._handle_message_stream(rpc_request) + + @app.get("/tasks/{task_id}") + async def get_task(task_id: str): + """获取任务状态""" + if task_id not in self.tasks: + raise HTTPException(status_code=404, detail="Task not found") + return self.tasks[task_id].model_dump() + + @app.get("/health") + async def health_check(): + """健康检查""" + return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()} + + async def _handle_message_send(self, request: A2ARequest) -> JSONResponse: + """处理 message/send 请求""" + params = request.params or {} + message_data = params.get("message", {}) + + # 提取用户消息文本 + user_text = "" + parts = message_data.get("parts", []) + for part in parts: + if part.get("kind") == "text": + user_text += part.get("text", "") + + if not user_text: + return JSONResponse({ + "jsonrpc": "2.0", + "id": request.id, + "error": { + "code": -32602, + "message": "Invalid params: no text content found" + } + }) + + # 创建任务 + task_id = uuid.uuid4().hex + context_id = params.get("contextId", uuid.uuid4().hex) + + task = A2ATask( + id=task_id, + contextId=context_id, + status=A2ATaskStatus(state="working") + ) + self.tasks[task_id] = task + + try: + # 调用Agent获取响应 + logger.info("处理消息", task_id=task_id, message_preview=user_text[:50]) + + response_text = await self.agent.chat( + message=user_text, + conversation_id=context_id + ) + + # 更新任务状态 + task.status = A2ATaskStatus(state="completed") + task.artifacts = [ + A2AArtifact( + name="response", + parts=[A2APart(kind="text", text=response_text)] + ) + ] + self.tasks[task_id] = task + + return JSONResponse({ + "jsonrpc": "2.0", + "id": request.id, + "result": task.model_dump() + }) + + except Exception as e: + logger.error("处理消息失败", error=str(e)) + task.status = A2ATaskStatus(state="failed", message=str(e)) + self.tasks[task_id] = task + + return JSONResponse({ + "jsonrpc": "2.0", + "id": request.id, + "error": { + "code": -32000, + "message": f"Agent error: {str(e)}" + } + }) + + async def _handle_message_stream(self, request: A2ARequest) -> StreamingResponse: + """处理 message/stream 请求 (SSE)""" + params = request.params or {} + message_data = params.get("message", {}) + + # 提取用户消息 + user_text = "" + parts = message_data.get("parts", []) + for part in parts: + if part.get("kind") == "text": + user_text += part.get("text", "") + + task_id = uuid.uuid4().hex + context_id = params.get("contextId", uuid.uuid4().hex) + + async def event_generator() -> AsyncGenerator[str, None]: + """生成SSE事件流""" + # 发送任务开始事件 + start_event = { + "kind": "task-start", + "taskId": task_id, + "contextId": context_id + } + yield f"data: {json.dumps(start_event)}\n\n" + + try: + # 获取流式响应 + stream = await self.agent.chat( + message=user_text, + conversation_id=context_id, + stream=True + ) + + full_response = "" + async for chunk in stream: + full_response += chunk + # 发送文本增量事件 + delta_event = { + "kind": "artifact-delta", + "taskId": task_id, + "contextId": context_id, + "data": { + "kind": "text", + "text": chunk + } + } + yield f"data: {json.dumps(delta_event)}\n\n" + + # 发送完成事件 + complete_event = { + "kind": "task-complete", + "taskId": task_id, + "contextId": context_id, + "data": { + "status": "completed", + "artifacts": [{ + "name": "response", + "parts": [{"kind": "text", "text": full_response}] + }] + } + } + yield f"data: {json.dumps(complete_event)}\n\n" + + except Exception as e: + # 发送错误事件 + error_event = { + "kind": "task-error", + "taskId": task_id, + "contextId": context_id, + "data": { + "error": str(e) + } + } + yield f"data: {json.dumps(error_event)}\n\n" + + return StreamingResponse( + event_generator(), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no" + } + ) + + def run(self, host: Optional[str] = None, port: Optional[int] = None): + """运行服务器""" + import uvicorn + + host = host or self.agent_config.host + port = port or self.agent_config.port + + logger.info(f"启动A2A Agent服务", host=host, port=port) + uvicorn.run(self.app, host=host, port=port) + + +def create_app(api_key: Optional[str] = None, model: Optional[str] = None) -> FastAPI: + """ + 创建FastAPI应用（用于uvicorn启动） + + 使用方式: + uvicorn a2a_server:app --host 0.0.0.0 --port 8080 + + 或设置环境变量后: + export LITELLM_API_KEY="your-key" + export LITELLM_MODEL="your-model" + uvicorn a2a_server:app --host 0.0.0.0 --port 8080 + """ + import os + api_key = api_key or os.getenv("LITELLM_API_KEY") + model = model or os.getenv("LITELLM_MODEL") + + server = A2AAgentServer(api_key=api_key, model=model) + return server.app + + +# uvicorn 启动入口 +# 环境变量: LITELLM_API_KEY, LITELLM_MODEL +app = create_app() diff --git a/search_agent/a2a-ai-search-agent/agent.py b/search_agent/a2a-ai-search-agent/agent.py new file mode 100644 index 0000000..f190c64 --- /dev/null +++ b/search_agent/a2a-ai-search-agent/agent.py @@ -0,0 +1,333 @@ +""" +LiteLLM Agent 核心模块 +基于LiteLLM框架的Agent实现，支持A2A协议 +""" + +import asyncio +import json +import uuid +from typing import AsyncGenerator, Optional, Dict, Any, List +from dataclasses import dataclass, field +from datetime import datetime + +import httpx +import structlog + +from config import LiteLLMConfig, AgentConfig, get_config + +# 配置日志 +logger = structlog.get_logger() + + +@dataclass +class Message: + """消息数据结构""" + role: str # user, assistant, system + content: str + message_id: str = field(default_factory=lambda: uuid.uuid4().hex) + timestamp: datetime = field(default_factory=datetime.now) + metadata: Dict[str, Any] = field(default_factory=dict) + + +@dataclass +class Conversation: + """对话上下文""" + conversation_id: str = field(default_factory=lambda: uuid.uuid4().hex) + messages: List[Message] = field(default_factory=list) + created_at: datetime = field(default_factory=datetime.now) + + def add_message(self, role: str, content: str) -> Message: + """添加消息到对话""" + msg = Message(role=role, content=content) + self.messages.append(msg) + return msg + + def to_openai_format(self) -> List[Dict[str, str]]: + """转换为OpenAI格式的消息列表""" + return [{"role": m.role, "content": m.content} for m in self.messages] + + +class LiteLLMAgent: + """ + 基于LiteLLM的Agent实现 + + 支持功能: + - 多轮对话 + - 流式响应 + - 工具调用 + - A2A协议兼容 + """ + + def __init__( + self, + api_key: Optional[str] = None, + model: Optional[str] = None, + litellm_config: Optional[LiteLLMConfig] = None, + agent_config: Optional[AgentConfig] = None + ): + """ + 初始化Agent + + Args: + api_key: LiteLLM API密钥 + model: 模型名称 + litellm_config: LiteLLM配置对象 + agent_config: Agent配置对象 + """ + if litellm_config: + self.llm_config = litellm_config + else: + self.llm_config = LiteLLMConfig(api_key=api_key, model=model) + + if agent_config: + self.agent_config = agent_config + else: + self.agent_config = AgentConfig() + + # 验证配置 + self.llm_config.validate() + + # HTTP客户端 + self._client: Optional[httpx.AsyncClient] = None + + # 对话管理 + self.conversations: Dict[str, Conversation] = {} + + # 工具注册 + self.tools: Dict[str, callable] = {} + + logger.info( + "Agent初始化完成", + agent_name=self.agent_config.name, + model=self.llm_config.model, + base_url=self.llm_config.base_url + ) + + async def _get_client(self) -> httpx.AsyncClient: + """获取或创建HTTP客户端""" + if self._client is None or self._client.is_closed: + self._client = httpx.AsyncClient( + timeout=httpx.Timeout(self.llm_config.timeout), + headers={ + "Authorization": f"Bearer {self.llm_config.api_key}", + "Content-Type": "application/json" + } + ) + return self._client + + async def close(self): + """关闭资源""" + if self._client and not self._client.is_closed: + await self._client.aclose() + + def register_tool(self, name: str, func: callable, description: str = ""): + """注册工具函数""" + self.tools[name] = { + "function": func, + "description": description + } + logger.info(f"注册工具: {name}") + + def get_or_create_conversation(self, conversation_id: Optional[str] = None) -> Conversation: + """获取或创建对话""" + if conversation_id and conversation_id in self.conversations: + return self.conversations[conversation_id] + + conv = Conversation(conversation_id=conversation_id or uuid.uuid4().hex) + # 添加系统提示 + conv.add_message("system", self.agent_config.system_prompt) + self.conversations[conv.conversation_id] = conv + return conv + + async def chat( + self, + message: str, + conversation_id: Optional[str] = None, + stream: bool = False + ) -> str | AsyncGenerator[str, None]: + """ + 发送消息并获取回复 + + Args: + message: 用户消息 + conversation_id: 对话ID（用于多轮对话） + stream: 是否流式响应 + + Returns: + 如果stream=False，返回完整回复字符串 + 如果stream=True，返回异步生成器 + """ + # 获取对话上下文 + conversation = self.get_or_create_conversation(conversation_id) + conversation.add_message("user", message) + + if stream: + return self._stream_chat(conversation) + else: + return await self._simple_chat(conversation) + + async def _simple_chat(self, conversation: Conversation) -> str: + """非流式对话""" + client = await self._get_client() + + request_body = { + "model": self.llm_config.model, + "messages": conversation.to_openai_format(), + "temperature": self.llm_config.temperature, + "max_tokens": self.llm_config.max_tokens + } + + logger.debug("发送请求", endpoint=self.llm_config.chat_endpoint) + + try: + response = await client.post( + self.llm_config.chat_endpoint, + json=request_body + ) + response.raise_for_status() + + result = response.json() + assistant_message = result["choices"][0]["message"]["content"] + + # 保存助手回复到对话 + conversation.add_message("assistant", assistant_message) + + logger.info("收到回复", length=len(assistant_message)) + return assistant_message + + except httpx.HTTPStatusError as e: + logger.error("HTTP错误", status_code=e.response.status_code, detail=e.response.text) + raise + except Exception as e: + logger.error("请求失败", error=str(e)) + raise + + async def _stream_chat(self, conversation: Conversation) -> AsyncGenerator[str, None]: + """流式对话""" + client = await self._get_client() + + request_body = { + "model": self.llm_config.model, + "messages": conversation.to_openai_format(), + "temperature": self.llm_config.temperature, + "max_tokens": self.llm_config.max_tokens, + "stream": True + } + + full_response = "" + + try: + async with client.stream( + "POST", + self.llm_config.chat_endpoint, + json=request_body + ) as response: + response.raise_for_status() + + async for line in response.aiter_lines(): + if line.startswith("data: "): + data = line[6:] + if data == "[DONE]": + break + + try: + chunk = json.loads(data) + delta = chunk.get("choices", [{}])[0].get("delta", {}) + content = delta.get("content", "") + if content: + full_response += content + yield content + except json.JSONDecodeError: + continue + + # 保存完整回复到对话 + conversation.add_message("assistant", full_response) + + except Exception as e: + logger.error("流式请求失败", error=str(e)) + raise + + async def invoke_tool(self, tool_name: str, **kwargs) -> Any: + """调用注册的工具""" + if tool_name not in self.tools: + raise ValueError(f"未找到工具: {tool_name}") + + tool = self.tools[tool_name] + func = tool["function"] + + logger.info(f"调用工具: {tool_name}", kwargs=kwargs) + + if asyncio.iscoroutinefunction(func): + return await func(**kwargs) + else: + return func(**kwargs) + + +# 示例工具函数 +def tool_get_current_time() -> str: + """获取当前时间""" + return datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + +def tool_calculate(expression: str) -> str: + """计算数学表达式""" + try: + # 安全的数学计算 + allowed_chars = set("0123456789+-*/.() ") + if not all(c in allowed_chars for c in expression): + return "错误: 不支持的字符" + result = eval(expression) + return str(result) + except Exception as e: + return f"计算错误: {str(e)}" + + +async def main(): + """示例用法""" + import os + + # 从环境变量或命令行获取配置 + api_key = os.getenv("LITELLM_API_KEY") + model = os.getenv("LITELLM_MODEL", "gpt-4") + + if not api_key: + print("请设置 LITELLM_API_KEY 环境变量") + print("示例: export LITELLM_API_KEY='your-api-key'") + return + + # 创建Agent + agent = LiteLLMAgent(api_key=api_key, model=model) + + # 注册工具 + agent.register_tool("get_time", tool_get_current_time, "获取当前时间") + agent.register_tool("calculate", tool_calculate, "计算数学表达式") + + try: + # 简单对话示例 + print("\n=== 简单对话 ===") + response = await agent.chat("你好，请介绍一下你自己") + print(f"Agent: {response}") + + # 多轮对话示例 + print("\n=== 多轮对话 ===") + conv_id = "test-conversation" + response1 = await agent.chat("我的名字叫张三", conversation_id=conv_id) + print(f"Agent: {response1}") + + response2 = await agent.chat("你还记得我的名字吗？", conversation_id=conv_id) + print(f"Agent: {response2}") + + # 流式响应示例 + print("\n=== 流式响应 ===") + print("Agent: ", end="", flush=True) + stream = await agent.chat("请用3句话描述Python编程语言", stream=True) + async for chunk in stream: + print(chunk, end="", flush=True) + print() + + finally: + await agent.close() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/search_agent/a2a-ai-search-agent/config.py b/search_agent/a2a-ai-search-agent/config.py new file mode 100644 index 0000000..44f5241 --- /dev/null +++ b/search_agent/a2a-ai-search-agent/config.py @@ -0,0 +1,141 @@ +""" +LiteLLM Agent 配置模块 +支持用户传入密钥和模型名称 +""" + +import os +from dataclasses import dataclass, field +from typing import Optional +from dotenv import load_dotenv + +# 加载环境变量 +load_dotenv() + + +@dataclass +class LiteLLMConfig: + """LiteLLM 配置""" + # 基础URL - 用户提供的LiteLLM服务地址 + base_url: str = "https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io" + + # 完整的chat completions端点 + chat_endpoint: str = field(init=False) + + # API密钥 - 需要用户传入 + api_key: Optional[str] = None + + # 模型名称 - 需要用户传入 + model: Optional[str] = None + + # 请求超时时间(秒) + timeout: int = 120 + + # 最大重试次数 + max_retries: int = 3 + + # 温度参数 + temperature: float = 0.7 + + # 最大token数 + max_tokens: int = 4096 + + def __post_init__(self): + self.chat_endpoint = f"{self.base_url}/chat/completions" + + # 从环境变量读取（如果未直接提供） + if self.api_key is None: + self.api_key = os.getenv("LITELLM_API_KEY") + if self.model is None: + self.model = os.getenv("LITELLM_MODEL", "gpt-4") + + def validate(self) -> bool: + """验证配置是否完整""" + if not self.api_key: + raise ValueError("API密钥未设置! 请设置 LITELLM_API_KEY 环境变量或直接传入 api_key") + if not self.model: + raise ValueError("模型名称未设置! 请设置 LITELLM_MODEL 环境变量或直接传入 model") + return True + + +@dataclass +class AgentConfig: + """Agent 配置""" + # Agent名称 + name: str = "xiaohei-agent" + + # Agent描述 + description: str = "一个基于LiteLLM的智能Agent，支持A2A协议" + + # Agent版本 + version: str = "1.0.0" + + # 服务端口 + port: int = 8080 + + # 服务主机 + host: str = "0.0.0.0" + + # 是否启用流式响应 + enable_streaming: bool = True + + # 系统提示词 + system_prompt: str = """你是小黑Agent，一个智能助手。 +你可以帮助用户完成各种任务，包括： +- 回答问题 +- 代码编写和解释 +- 文档分析 +- 任务规划 + +请用中文回答用户的问题，保持友好和专业。""" + + +@dataclass +class A2AConfig: + """A2A协议配置""" + # A2A协议版本 + protocol_version: str = "1.0" + + # Agent Card配置 + agent_card: dict = field(default_factory=lambda: { + "name": "xiaohei-agent", + "description": "基于LiteLLM的智能Agent，支持A2A协议通信", + "version": "1.0.0", + "capabilities": { + "text": True, + "streaming": True, + "push_notifications": False + }, + "skills": [ + { + "id": "general-assistant", + "name": "通用助手", + "description": "回答问题、提供建议、协助任务" + }, + { + "id": "code-helper", + "name": "代码助手", + "description": "编写、解释和调试代码" + } + ] + }) + + +def get_config( + api_key: Optional[str] = None, + model: Optional[str] = None +) -> tuple[LiteLLMConfig, AgentConfig, A2AConfig]: + """ + 获取完整配置 + + Args: + api_key: LiteLLM API密钥（可选，也可通过环境变量设置） + model: 模型名称（可选，也可通过环境变量设置） + + Returns: + (LiteLLMConfig, AgentConfig, A2AConfig) 配置元组 + """ + litellm_config = LiteLLMConfig(api_key=api_key, model=model) + agent_config = AgentConfig() + a2a_config = A2AConfig() + + return litellm_config, agent_config, a2a_config diff --git a/search_agent/a2a-ai-search-agent/deploy.sh b/search_agent/a2a-ai-search-agent/deploy.sh new file mode 100755 index 0000000..effb8d2 --- /dev/null +++ b/search_agent/a2a-ai-search-agent/deploy.sh @@ -0,0 +1,27 @@ +#!/bin/bash +# 部署小黑Agent到AKS + +echo "开始部署小黑Agent到AKS..." + +# 应用ConfigMap +kubectl apply -f k8s/configmap.yaml + +# 应用Secret +kubectl apply -f k8s/secret.yaml + +# 应用Deployment +kubectl apply -f k8s/deployment.yaml + +# 应用Service +kubectl apply -f k8s/service.yaml + +echo "部署完成！" +echo "" +echo "查看Pod状态:" +echo " kubectl get pods -l app=xiaohei-agent" +echo "" +echo "查看Service状态:" +echo " kubectl get svc xiaohei-agent-service" +echo "" +echo "查看外网IP (可能需要几分钟):" +echo " kubectl get svc xiaohei-agent-service -w" diff --git a/search_agent/a2a-ai-search-agent/k8s/configmap.yaml b/search_agent/a2a-ai-search-agent/k8s/configmap.yaml new file mode 100644 index 0000000..3274ec3 --- /dev/null +++ b/search_agent/a2a-ai-search-agent/k8s/configmap.yaml @@ -0,0 +1,11 @@ +# 小黑Agent ConfigMap +# 存储非敏感配置 + +apiVersion: v1 +kind: ConfigMap +metadata: + name: xiaohei-config + labels: + app: xiaohei-agent +data: + litellm-model: "taiji/gpt-4o-mini" diff --git a/search_agent/a2a-ai-search-agent/k8s/deployment.yaml b/search_agent/a2a-ai-search-agent/k8s/deployment.yaml new file mode 100644 index 0000000..df251d1 --- /dev/null +++ b/search_agent/a2a-ai-search-agent/k8s/deployment.yaml @@ -0,0 +1,70 @@ +# 小黑Agent Kubernetes 部署配置 +# 适用于 AKS (Azure Kubernetes Service) + +apiVersion: apps/v1 +kind: Deployment +metadata: + name: xiaohei-agent + labels: + app: xiaohei-agent + version: v1 +spec: + replicas: 2 + selector: + matchLabels: + app: xiaohei-agent + template: + metadata: + labels: + app: xiaohei-agent + version: v1 + spec: + containers: + - name: xiaohei-agent + image: agnettaiji.azurecr.io/xiaohei-agent:latest + imagePullPolicy: Always + ports: + - containerPort: 9000 + protocol: TCP + env: + - name: LITELLM_API_KEY + valueFrom: + secretKeyRef: + name: xiaohei-secrets + key: litellm-api-key + - name: LITELLM_MODEL + valueFrom: + configMapKeyRef: + name: xiaohei-config + key: litellm-model + resources: + requests: + cpu: "100m" + memory: "256Mi" + limits: + cpu: "500m" + memory: "512Mi" + # 存活探针 + livenessProbe: + httpGet: + path: /health + port: 9000 + initialDelaySeconds: 10 + periodSeconds: 30 + timeoutSeconds: 5 + failureThreshold: 3 + # 就绪探针 + readinessProbe: + httpGet: + path: /health + port: 9000 + initialDelaySeconds: 5 + periodSeconds: 10 + timeoutSeconds: 3 + failureThreshold: 3 + # 安全上下文 + securityContext: + runAsNonRoot: true + runAsUser: 1000 + # 优雅终止 + terminationGracePeriodSeconds: 30 diff --git a/search_agent/a2a-ai-search-agent/k8s/secret.yaml b/search_agent/a2a-ai-search-agent/k8s/secret.yaml new file mode 100644 index 0000000..9adfcaf --- /dev/null +++ b/search_agent/a2a-ai-search-agent/k8s/secret.yaml @@ -0,0 +1,13 @@ +# 小黑Agent Secret +# 存储敏感配置（API密钥等） +# 注意: 实际部署时请使用 Azure Key Vault 或 kubectl create secret + +apiVersion: v1 +kind: Secret +metadata: + name: xiaohei-secrets + labels: + app: xiaohei-agent +type: Opaque +stringData: + litellm-api-key: "sk-7tvnni908QJzuuUWedvx5w" diff --git a/search_agent/a2a-ai-search-agent/k8s/service.yaml b/search_agent/a2a-ai-search-agent/k8s/service.yaml new file mode 100644 index 0000000..5ce814d --- /dev/null +++ b/search_agent/a2a-ai-search-agent/k8s/service.yaml @@ -0,0 +1,18 @@ +# 小黑Agent Kubernetes Service +# 暴露服务供外部访问 + +apiVersion: v1 +kind: Service +metadata: + name: xiaohei-agent-service + labels: + app: xiaohei-agent +spec: + type: LoadBalancer # 使用Azure LoadBalancer暴露外网 + selector: + app: xiaohei-agent + ports: + - name: http + port: 80 + targetPort: 9000 + protocol: TCP diff --git a/search_agent/a2a-ai-search-agent/requirements.txt b/search_agent/a2a-ai-search-agent/requirements.txt new file mode 100644 index 0000000..8f44ea1 --- /dev/null +++ b/search_agent/a2a-ai-search-agent/requirements.txt @@ -0,0 +1,7 @@ +# LiteLLM Agent API服务依赖 +httpx>=0.27.0 +fastapi>=0.115.0 +uvicorn[standard]>=0.32.0 +pydantic>=2.0.0 +python-dotenv>=1.0.0 +structlog>=24.0.0