Python 快速开始Python quick start

This page targets agentscope 2.0.11. The goal is a working agent with tool calls, streaming and structured output in ten minutes. Every class name links to its full signature in the Python reference.

1. Setup

Python 3.11+ is required:

python --version
python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install "agentscope==2.0.11"

Export a model key. The examples use an OpenAI-compatible endpoint; DashScope, DeepSeek, Anthropic, Gemini and Ollama adapters work the same way:

export OPENAI_API_KEY="sk-..."    # Windows PowerShell: $env:OPENAI_API_KEY="sk-..."

2. Create a model

A model is credential + model name + parameters. Credentials are separate objects so they can be reused across models:

from agentscope.credential import OpenAICredential
from agentscope.model import OpenAIChatModel

model = OpenAIChatModel(
    credential=OpenAICredential(api_key="sk-..."),
    model="gpt-4o-mini",
    stream=True,
    context_size=128_000,
)

agentscope.credential also provides DashScopeCredential, DeepSeekCredential, AnthropicCredential, GeminiCredential, OllamaCredential and others. The adapter picks the matching formatter; pass formatter= explicitly to customize the protocol.

3. Define a tool

A tool is a plain Python function. The framework derives a JSON Schema from the signature and the Google-style docstring:

from agentscope.tool import FunctionTool, Toolkit

def get_weather(city: str) -> str:
    """Return the current weather for a city.

    Args:
        city: The city name, for example "Hangzhou".
    """
    # Call a real weather API in production
    return f"{city}: sunny, 24C"

toolkit = Toolkit(tools=[FunctionTool(get_weather)])

Calls are validated and permission-checked before execution. When human confirmation is required, the event stream carries a confirmation request. For parallel execution, construct the tool with FunctionTool(is_concurrency_safe=True).

4. Build the agent and chat

import asyncio

from agentscope.agent import Agent, ReActConfig
from agentscope.message import UserMsg


async def main() -> None:
    agent = Agent(
        name="assistant",
        system_prompt="You are a concise assistant. Use tools when needed.",
        model=model,
        toolkit=toolkit,
        react_config=ReActConfig(max_iters=10),
    )

    reply = await agent.reply(UserMsg("user", "What's the weather in Hangzhou?"))
    print(reply.get_text_content())


asyncio.run(main())

reply runs the whole ReAct loop: the model decides to call get_weather, the result is written back to the context, and the model produces the final answer. The returned Msg carries content, usage, finished_reason and other metadata.

5. Streaming

Use reply_stream for incremental rendering. Events cover reply start, model calls, text deltas, thinking deltas, tool calls, tool results and reply end:

from agentscope.event import TextBlockDeltaEvent

async for event in agent.reply_stream(UserMsg("user", "Explain ReAct in two sentences.")):
    if isinstance(event, TextBlockDeltaEvent):
        print(event.delta, end="", flush=True)

Pass yield_final_msg=True to receive the final Msg at the end of the stream.

6. Structured output

Pass a Pydantic model to constrain the result:

from pydantic import BaseModel


class WeatherReport(BaseModel):
    city: str
    temperature_c: float
    condition: str


reply = await agent.reply(
    UserMsg("user", "Report the weather in Hangzhou as structured data."),
    structured_schema=WeatherReport,
)
print(reply.structured_output)   # dict matching WeatherReport

Structured output is implemented as a dedicated tool inside the loop, so it still obeys tool permissions and the iteration limit. If the model cannot produce a valid value, finished_reason records the failure.

7. Multi-turn state

Pass the same AgentState to keep context across reply calls:

from agentscope.state import AgentState

state = AgentState(session_id="demo-session")
agent = Agent(
    name="assistant",
    system_prompt="You are a helpful assistant.",
    model=model,
    state=state,
)

await agent.reply(UserMsg("user", "My name is Ada."))
reply = await agent.reply(UserMsg("user", "What is my name?"))

To persist, serialize the state and store it in your database, Redis or object storage:

payload = state.model_dump_json()
# save(payload)
# state = AgentState.model_validate_json(payload)

When the context grows too long, ContextConfig triggers compression. Compression produces a structured summary and preserves recent messages, tool calls and task state; it is visible in the event stream.

8. Add RAG

RAG has four replaceable parts: parsers, chunkers, embedding models and vector stores. A local Milvus Lite setup:

from agentscope.embedding import OpenAIEmbeddingModel
from agentscope.rag import KnowledgeBase, MilvusLiteStore

embedding = OpenAIEmbeddingModel(
    credential=OpenAICredential(api_key="sk-..."),
    model="text-embedding-3-small",
    dimensions=1536,
)
store = MilvusLiteStore(uri="./knowledge.db")

kb = KnowledgeBase(
    name="handbook",
    description="Internal engineering handbook",
    embedding_model=embedding,
    vector_store=store,
    collection="handbook",
)

# Parse documents first (PDFParser / WordParser / TextParser, ...), then insert_document(chunks)
results = await kb.search(["deployment checklist"], top_k=5)
for item in results:
    print(item)

Wrap retrieval as a tool, or use RAGMiddleware to inject results before the model call. In production, swap MilvusLiteStore for QdrantStore, MongoDBStore or ElasticsearchStore.

9. Next steps

• See Agent for the full constructor and the input variants of reply_stream.
• See Toolkit for tool groups, MCP and skill registration.
• See KnowledgeBase for document management and retrieval.
• For multiple agents, read TeamPipeline.
• For service deployment, browse the storage, channel and message-bus APIs under agentscope.app.