ACP协议实战指南：从零构建你的AI智能体团队（下篇）

import os
import asyncio
from collections.abc import Iterator
from datetime import datetime
from acp_sdk import Message, MessagePart
from acp_sdk.server import Context, Server
from crewai import Agent, Crew, Task, Process
from crewai.llm import LLM
from crewai.tools import tool

# 设置LLM，这里用Gemini，你也可以换成其他的
GEMINI_API_KEY = os.getenv('GEMINI_API_KEY')
if not GEMINI_API_KEY:
    raise ValueError("需要设置GEMINI_API_KEY环境变量")

llm = LLM(model='gemini/gemini-2.0-flash-exp',
         api_key=GEMINI_API_KEY,
         temperature=0.7)  # 温度设高一点，让诗歌更有创意

server = Server()

# 全局会话ID管理
_current_session_id = None

def generate_unique_session_id():
    """生成唯一的会话ID"""
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    import uuid
    short_uuid = uuid.uuid4().hex[:6]
    return f"{timestamp}_{short_uuid}"

@tool("Text to Speech")
def text_to_speech_tool(text: str) -> str:
    """将文本转换为语音文件"""
    global _current_session_id
    os.makedirs("poem", exist_ok=True)
    output_path = f"poem/speech_{_current_session_id}.wav"
    
    # 这里简化处理，实际可以集成真实的TTS服务
    print(f"🗣️ 正在生成语音: {text[:50]}...")
    
    # 创建占位文件（实际应用中会调用TTS API）
    with open(output_path, 'w', encoding='utf-8') as f:
        f.write(f"TTS音频文件占位符\n文本内容: {text}")
    
    print(f"✅ 语音文件已生成: {output_path}")
    return output_path

@tool("Music Generation")
def music_generation_tool(prompt: str) -> str:
    """根据提示生成背景音乐"""
    global _current_session_id
    os.makedirs("poem", exist_ok=True)
    output_path = f"poem/music_{_current_session_id}.wav"
    
    print(f"🎵 正在生成音乐: {prompt}")
    
    # 创建占位文件（实际应用中会调用音乐生成模型）
    with open(output_path, 'w', encoding='utf-8') as f:
        f.write(f"音乐文件占位符\n音乐风格: {prompt}")
    
    print(f"✅ 音乐文件已生成: {output_path}")
    return output_path

@server.agent()
def poetry_agent(input: list[Message], context: Context) -> Iterator:
    """多智能体诗歌创作系统"""
    global _current_session_id
    
    theme = str(input[-1].parts[0].content) if input else ""
    if not theme:
        yield MessagePart(content="请提供一个诗歌主题。")
        return

    # 生成会话ID
    _current_session_id = generate_unique_session_id()
    
    try:
        # 定义专业智能体团队
        poet = Agent(
            llm=llm,
            role="诗人",
            goal="根据主题 {theme} 创作一首优美的短诗（4-8行）。如果主题是中文，请用中文创作；如果是英文，请用英文创作。",
            backstory="你是一位精通中英文的专业诗人，擅长创作富有感染力的短诗。",
            verbose=True,
        )

        voice_artist = Agent(
            llm=llm,
            role="配音师",
            goal="将诗歌转换为清晰动人的语音文件。",
            backstory="你是一位专业的配音演员，能够将文字转化为富有情感的声音。",
            verbose=True,
            tools=[text_to_speech_tool],
        )

        musician = Agent(
            llm=llm,
            role="音乐家",
            goal="为诗歌创作合适的背景音乐。",
            backstory="你是一位才华横溢的作曲家，专长于创作与诗歌情境匹配的音乐。",
            verbose=True,
            tools=[music_generation_tool],
        )

        # 定义创作任务
        write_poem_task = Task(
            description="根据主题 '{theme}' 创作一首短诗（4-8行）。保持语言与主题一致。",
            expected_output="一首4-8行的优美诗歌",
            agent=poet,
        )

        speak_poem_task = Task(
            description="将诗歌转换为语音文件。",
            expected_output="语音文件路径",
            agent=voice_artist,
            context=[write_poem_task]
        )

        create_music_task = Task(
            description="为诗歌创作适合的背景音乐。",
            expected_output="音乐文件路径",
            agent=musician,
            context=[write_poem_task]
        )

        # 创建并运行智能体团队
        crew = Crew(
            agents=[poet, voice_artist, musician],
            tasks=[write_poem_task, speak_poem_task, create_music_task],
            process=Process.sequential,
            verbose=True,
        )

        result = crew.kickoff(inputs={'theme': theme})

        # 处理结果
        poem_content = None
        speech_file = None
        music_file = None
        
        # 提取任务输出
        if hasattr(result, 'tasks_output'):
            for task_output in result.tasks_output:
                output_text = str(task_output.raw) if hasattr(task_output, 'raw') else str(task_output)
                
                # 提取诗歌内容
                if poem_content is None and 'poem/' not in output_text and '.wav' not in output_text:
                    if len(output_text.strip()) > 20:
                        poem_content = output_text.strip()
                
                # 提取文件路径
                if 'poem/speech_' in output_text:
                    speech_file = f"poem/speech_{_current_session_id}.wav"
                elif 'poem/music_' in output_text:
                    music_file = f"poem/music_{_current_session_id}.wav"

        # 保存诗歌文本
        if poem_content:
            poem_file_path = f"poem/poem_text_{_current_session_id}.txt"
            with open(poem_file_path, 'w', encoding='utf-8') as f:
                f.write(f"主题: {theme}\n")
                f.write(f"创作时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
                f.write(f"{'='*50}\n")
                f.write(poem_content)
            
            yield MessagePart(content=f"🎉 诗歌创作完成！会话ID: {_current_session_id}\n\n"
                                    f"📝 诗歌: {poem_file_path}\n"
                                    f"🗣️ 配音: {speech_file or '生成中...'}\n"
                                    f"🎵 音乐: {music_file or '生成中...'}\n\n"
                                    f"诗歌内容预览:\n{'-'*30}\n{poem_content[:200]}...")
        else:
            yield MessagePart(content=f"❌ 诗歌创作失败，会话ID: {_current_session_id}")
            
    except Exception as e:
        yield MessagePart(content=f"❌ 创作过程出错: {str(e)}")

if __name__ == "__main__":
    print("🎭 启动诗歌创作智能体团队...")
    print("📍 服务地址: http://localhost:8000")
    server.run(host="localhost", port=8000)

import asyncio
from acp_sdk.client import Client
from acp_sdk.models import Message, MessagePart

async def create_poetry():
    """调用诗歌创作智能体"""
    print("连接到诗歌创作服务...")
    
    async with Client(base_url="http://localhost:8000") as client:
        # 试试不同的主题
        themes = ["八月太热了"]
        
        for theme in themes:
            print(f"\n创作主题: {theme}")
            print("-" * 50)
            
            run = await client.run_sync(
                agent="poetry_agent",
                input=[Message(parts=[
                    MessagePart(content=theme, content_type="text/plain")
                ])]
            )
            
            for message in run.output:
                for part in message.parts:
                    print(part.content)
            
            print("-" * 50)

if __name__ == "__main__":
    asyncio.run(create_poetry())

# 它能访问这些文档来学习最新的API
documentation_sources = [
    "ACP协议文档",
    "LangGraph框架文档", 
    "BeeAI平台文档"
]

@server.agent()
async def acp_agent_generator(input: list[Message], context: Context):
    """这就是核心的生成逻辑"""
    user_request = input[-1].parts[0].content
    
    # 先理解用户要什么
    requirements = analyze_requirements(user_request)
    
    # 查文档找最佳实践
    docs = await query_documentation(requirements)
    
    # 生成代码
    generated_code = await generate_agent_code(requirements, docs)
    
    yield generated_code