agentscope-runtime

import os import uvicorn from contextlib import asynccontextmanager from fastapi import FastAPI from agentscope.agent import ReActAgent from agentscope.model import DashScopeChatModel from agentscope.formatter import DashScopeChatFormatter from agentscope.memory import InMemoryMemory from agentscope.session import RedisSession from agentscope_runtime.engine import AgentApp from agentscope_runtime.engine.schemas.agent_schemas import AgentRequest from agentscope.pipeline import stream_printing_messages import fakeredis # For local development/testing # Configure API key (replace with your actual key or other LLM provider config) # os.environ['DASHSCOPE_API_KEY'] = 'YOUR_API_KEY_HERE' @asynccontextmanager async def lifespan(app: FastAPI): # Initialize Session manager (using fakeredis for demonstration) fake_redis = fakeredis.aioredis.FakeRedis(decode_responses=True) app.state.session = RedisSession(connection_pool=fake_redis.connection_pool) print("\n🚀 AgentApp starting up...") yield print("\n🛑 AgentApp shutting down...") agent_app = AgentApp(lifespan=lifespan) @agent_app.query(framework="agentscope") async def query_agent(msgs, request: AgentRequest = None, **kwargs): # Example agent setup - customize with your model, tools, and memory dashscope_api_key = os.environ.get('DASHSCOPE_API_KEY', 'sk-example') # Use actual key in production if not dashscope_api_key or dashscope_api_key == 'sk-example': print("Warning: DASHSCOPE_API_KEY not set. Using dummy key. Model interactions may fail.") model = DashScopeChatModel( model_name="qwen-plus", # Or your preferred model api_key=dashscope_api_key, # Other model parameters like `temperature`, `max_tokens` can be set here ) agent = ReActAgent( name="SimpleAgent", model=model, memory=InMemoryMemory(), # Or RedisSession for persistence formatter=DashScopeChatFormatter(), ) # Load session context (if using RedisSession in app.state.session) if hasattr(request, 'session_id') and app.state.session: session_id = request.session_id # In a real scenario, you'd load context using session_id # For this simple example, we'll just acknowledge it. print(f"Processing request for session_id: {session_id}") # Stream responses async for msg, last in stream_printing_messages(agent.reply(msgs, **kwargs)): yield msg, last # To run this: save as e.g., `main.py` and run `uvicorn main:agent_app --reload --port 8090` # Test with curl: # curl -N -X POST "http://localhost:8090/process" \ # -H "Content-Type: application/json" \ # -d '{ "input": [ { "role": "user", "content": [ { "type": "text", "text": "What is the capital of France?" } ] } ], "session_id": "session_1" }'