chenchen/aks_agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add A2A AI Search Agent to root directory

Browse Source
This commit is contained in:
zhanggangyong
2026-01-15 15:16:51 +00:00
parent b5204db28e
commit a946e7d288
12 changed files with 1336 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
a2a-ai-search-agent/.gitignore
+24
View File
@@ -0,0 +1,24 @@
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
venv/
.venv/
ENV/
env/
# IDE
.vscode/
.idea/
*.swp
*.swo
# 环境变量
.env
*.log
# 系统文件
.DS_Store
Thumbs.db
a2a-ai-search-agent/Dockerfile
+24
View File
@@ -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"]
a2a-ai-search-agent/README.md
+181
View File
@@ -0,0 +1,181 @@
# A2A AI Search Agent
基于 LiteLLM 和 A2A (Agent2Agent) 协议的智能 Agent API 服务。
## 🚀 功能特性
- ✅ **A2A 协议兼容** - 完全支持 Google Agent2Agent 协议
- ✅ **多轮对话** - 维护对话上下文，支持连续对话
- ✅ **流式响应** - 支持 SSE 流式输出
- ✅ **LiteLLM 集成** - 统一的 LLM 调用接口
- ✅ **生产就绪** - 适用于 AKS (Azure Kubernetes Service) 部署
## 📁 项目结构
```
a2a-ai-search-agent/
├── config.py # 配置管理
├── agent.py # Agent 核心逻辑
├── a2a_server.py # A2A 协议 API 服务
├── requirements.txt # Python 依赖
├── Dockerfile # Docker 镜像
├── deploy.sh # 部署脚本
└── k8s/ # Kubernetes 配置
├── deployment.yaml
├── service.yaml
├── configmap.yaml
└── secret.yaml
```
## 🔧 配置
**环境变量：**
- `LITELLM_API_KEY` - LiteLLM API 密钥（必需）
- `LITELLM_MODEL` - 模型名称（如: taiji/gpt-4o-mini）
**LiteLLM 服务地址：**
```
https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io
```
## 🐳 Docker 部署
### 构建镜像
```bash
docker build -t a2a-ai-search-agent:latest .
```
### 运行容器
```bash
docker run -d \
-p 9000:9000 \
-e LITELLM_API_KEY="your-api-key" \
-e LITELLM_MODEL="taiji/gpt-4o-mini" \
a2a-ai-search-agent:latest
```
## ☸️ Kubernetes 部署
### 推送到 ACR
```bash
# 登录 ACR
docker login agnettaiji.azurecr.io -u agnettaiji -p 'YOUR_PASSWORD'
# 打标签
docker tag a2a-ai-search-agent:latest agnettaiji.azurecr.io/xiaohei-agent:latest
# 推送
docker push agnettaiji.azurecr.io/xiaohei-agent:latest
```
### 部署到 AKS
```bash
# 方式1: 使用脚本
./deploy.sh
# 方式2: 手动部署
kubectl apply -f k8s/configmap.yaml
kubectl apply -f k8s/secret.yaml
kubectl apply -f k8s/deployment.yaml
kubectl apply -f k8s/service.yaml
```
### 查看部署状态
```bash
# 查看 Pod
kubectl get pods -l app=xiaohei-agent
# 查看 Service 和外网 IP
kubectl get svc xiaohei-agent-service
# 实时监控外网 IP 分配
kubectl get svc xiaohei-agent-service -w
```
## 📡 API 端点
### 健康检查
```bash
GET /health
```
### Agent Card (A2A 发现)
```bash
GET /.well-known/agent.json
```
**响应示例：**
```json
{
"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": "回答问题、提供建议、协助完成各种任务"
}
]
}
```
### 发送消息 (非流式)
```bash
POST /message/send
Content-Type: application/json
{
"jsonrpc": "2.0",
"id": "request-1",
"method": "message/send",
"params": {
"message": {
"role": "user",
"parts": [{"kind": "text", "text": "你好"}],
"messageId": "msg-1"
}
}
}
```
### 发送消息 (流式)
```bash
POST /message/stream
Content-Type: application/json
{
"jsonrpc": "2.0",
"id": "request-1",
"method": "message/stream",
"params": {
"message": {
"role": "user",
"parts": [{"kind": "text", "text": "你好"}],
"messageId": "msg-1"
}
}
}
```
## 🔗 相关资源
- [A2A 协议规范](https://github.com/a2aproject/A2A)
- [LiteLLM 文档](https://docs.litellm.ai/)
- [Google Agent2Agent 介绍](https://developers.googleblog.com/en/a2a-a-new-era-of-agent-interoperability)
## 📄 许可证
MIT License
a2a-ai-search-agent/a2a_server.py
+487
View File
@@ -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()
a2a-ai-search-agent/agent.py
+333
View File
@@ -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())
a2a-ai-search-agent/config.py
+141
View File
@@ -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
a2a-ai-search-agent/deploy.sh
Executable
+27
View File
@@ -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"
a2a-ai-search-agent/k8s/configmap.yaml
+11
View File
@@ -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"
a2a-ai-search-agent/k8s/deployment.yaml
+70
View File
@@ -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
a2a-ai-search-agent/k8s/secret.yaml
+13
View File
@@ -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"
a2a-ai-search-agent/k8s/service.yaml
+18
View File
@@ -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
a2a-ai-search-agent/requirements.txt
+7
View File
@@ -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