0Pricing
AI Agents with LangChain & Autonomous Workflows · 강의

비동기 에이전트 실행

에이전트가 작업을 병렬로 수행하고 응답성을 높일 수 있도록 비동기 패턴을 구현하는 방법을 배웁니다.

비동기 에이전트 실행은(는) CoddyKit의 무료 AI Agents with LangChain & Autonomous Workflows 강의입니다. 이것은 4개 중 2번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 AI Agents with LangChain & Autonomous Workflows 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. AI Agents with LangChain & Autonomous Workflows 강의에는 총 4개의 강의가 포함되어 있습니다.

이 강의의 일부는 아직 번역되지 않았으며 영어로 표시됩니다.

Why Asynchronous Agents?

Imagine your AI agent needs to do several things at once: fetch data from two APIs, analyze text with an LLM, and then store a result. If it does these synchronously (one after another), it waits for each step to complete before starting the next.

This waiting can make your agent slow and unresponsive, especially when dealing with network calls or complex computations.

Synchronous vs. Asynchronous

Synchronous execution is like a single-lane road: only one car can pass at a time. If a car breaks down, all traffic stops.

  • Synchronous: Tasks run one by one.
  • Asynchronous: Tasks can start, pause while waiting for something (like an API response), and let other tasks run in the meantime. It's like a multi-lane highway or juggling multiple balls.

Asynchronous programming helps agents utilize idle time more effectively.

Python's Async/Await Keywords

Python uses the async and await keywords to enable asynchronous programming. Think of them as signals:

  • async def: Defines a function (called a coroutine) that can run asynchronously.
  • await: Pauses the current coroutine until the awaited task is complete, allowing other tasks to run.

Let's see a basic example:

import asyncio

async def say_hello():
    print("Hello ")
    await asyncio.sleep(1) # Simulate a delay
    print("World!")

async def main():
    await say_hello()

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

The asyncio Event Loop

Behind the scenes, Python's asyncio library manages how asynchronous tasks run. It uses an event loop.

  • The event loop constantly monitors tasks.
  • When an await statement pauses a task, the event loop switches to another ready task.
  • Once the awaited task is done, the event loop resumes the paused task.

This allows non-blocking operations, improving overall efficiency.

Async LLM Calls in LangChain

Many LangChain components, especially LLMs, offer asynchronous versions of their methods. For example, instead of .invoke(), you can often use .ainvoke() for an asynchronous call.

This is crucial for making your agent responsive when querying large language models, which can take time.

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

async def main():
    # Make sure you have your OpenAI API key set up
    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
    
    # Using .ainvoke() for an asynchronous call
    response = await llm.ainvoke([HumanMessage(content="What is the capital of Canada?")])
    print(response.content)

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

Running Multiple LLM Calls Concurrently

The real power of async shines when you need to make multiple LLM calls. You don't have to wait for each one to finish before starting the next.

Use asyncio.gather() to run several asynchronous tasks in parallel and collect their results.

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

async def main():
    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
    
    # Create multiple asynchronous LLM invocation tasks
    task1 = llm.ainvoke([HumanMessage(content="Tell me a fact about the sun.")])
    task2 = llm.ainvoke([HumanMessage(content="Tell me a fact about the moon.")])
    task3 = llm.ainvoke([HumanMessage(content="Tell me a fact about Earth.")])
    
    # Run all tasks concurrently and wait for them to complete
    results = await asyncio.gather(task1, task2, task3)
    
    for i, res in enumerate(results):
        print(f"Result {i+1}: {res.content}\n")

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

Asynchronous Tool Execution

Just like LLMs, your custom tools can also be asynchronous! If your tool performs I/O-bound operations (like fetching data from a database or an external API), making it asynchronous will significantly improve agent performance.

To create an async tool, implement the _arun method in your BaseTool subclass.

import asyncio
from langchain.tools import BaseTool

class AsyncWebSearchTool(BaseTool):
    name: str = "AsyncWebSearch"
    description: str = "Searches the web asynchronously for a query."

    async def _arun(self, query: str) -> str:
        # Simulate an asynchronous web search API call
        await asyncio.sleep(1.5) 
        return f"Results for '{query}': Found 5 articles."

    def _run(self, query: str) -> str:
        # Fallback for synchronous calls (optional but good practice)
        return f"Sync results for '{query}': Found 4 articles."

async def main():
    tool = AsyncWebSearchTool()
    result = await tool.arun("latest AI news")
    print(result)

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

Building Async Chains & Agents

When you combine asynchronous LLMs and tools into chains or agents, LangChain automatically leverages their async capabilities. Most LangChain runnables and chains also provide an .ainvoke() method.

This means you can build entire asynchronous workflows, allowing your agent to process complex tasks involving multiple steps and external calls much faster.

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableSequence

async def main():
    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
    prompt = ChatPromptTemplate.from_template("What is a unique fact about {animal}?")
    
    # Define a simple chain
    chain = prompt | llm
    
    # Invoke the chain asynchronously
    response = await chain.ainvoke({"animal": "platypus"})
    print(response.content)

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

Quick Check: Async Benefits

Let's check your understanding of asynchronous execution.

Recap: Mastering Async Agents

Great job! You've learned the fundamentals of asynchronous execution in Python and how to apply it to your LangChain agents.

  • Asynchronous programming (async/await) allows agents to perform tasks concurrently instead of waiting for each one.
  • This significantly improves responsiveness and throughput for I/O-bound operations.
  • LangChain's LLMs, tools, and chains often provide asynchronous methods (e.g., .ainvoke(), .arun()) to leverage this power.

By integrating async patterns, you can build more efficient and high-performing autonomous workflows!

자주 묻는 질문

“비동기 에이전트 실행” 강의는 무료인가요?

네 — “비동기 에이전트 실행” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Agents with LangChain & Autonomous Workflows 강의 전체를 잠금 해제할 수 있습니다. AI Agents with LangChain & Autonomous Workflows 강의에는 총 4개의 강의가 포함되어 있습니다.

“비동기 에이전트 실행”에서 뭘 배우나요?

에이전트가 작업을 병렬로 수행하고 응답성을 높일 수 있도록 비동기 패턴을 구현하는 방법을 배웁니다. 브라우저에서 직접 실행하는 실습 코드로 AI Agents with LangChain & Autonomous Workflows을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.

AI Agents with LangChain & Autonomous Workflows을(를) 시작하는 데 경험이 필요한가요?

사전 경험은 필요하지 않습니다. CoddyKit의 AI Agents with LangChain & Autonomous Workflows은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 2번째 강의입니다.

“비동기 에이전트 실행” 강의는 얼마나 걸리나요?

대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.

이 AI Agents with LangChain & Autonomous Workflows 강의에서 코드를 작성하고 실행할 수 있나요?

네. 모든 AI Agents with LangChain & Autonomous Workflows 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.

이 강의의 모든 강의

  1. 복잡한 작업 흐름 설계
  2. 비동기 에이전트 실행
  3. 오류 처리와 회복 탄력성
  4. 사람 참여형 승인
← AI Agents with LangChain & Autonomous Workflows(으)로 돌아가기